prop/lib/microwaveprop/pskr/client.ex

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

  Connects to `mqtt.pskreporter.info:1883` over plain TCP (the
  broker doesn't expose TLS), subscribes to a curated set of
  topics, and forwards every PUBLISH 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 reaches us — the
  alternative would be subscribing to `pskr/filter/v2/#` and
  dropping ~95% client-side.

  ## No external MQTT library

  We talk to the broker with `:gen_tcp` and the codec in
  `Microwaveprop.Pskr.Mqtt`. emqtt's transitive `quicer` dep was a
  CMake/native-build mess in our slim Debian builder image; the
  alternative pure-Elixir client (tortoise311) bundles a
  gen_state_machine + retry/inflight stack we don't use because
  we're subscribe-only at QoS 0. Implementing the wire protocol
  inline is ~150 lines of binary parsing — less code than the
  dependency-management overhead of either choice.

  ## 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, which
  libcluster gives us). The losing nodes stay in `:standby` mode:
  no socket, no DB writes, just an idle process watching for the
  leader's `nodedown` to re-run election.

  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 in `Pskr.Aggregator` 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. Microwave bands are listed too but spots
  there are vanishingly rare (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
  alias Microwaveprop.Pskr.Mqtt

  require Logger

  @broker_host ~c"mqtt.pskreporter.info"
  @broker_port 1883
  @keepalive_seconds 60
  @keepalive_ms @keepalive_seconds * 1000

  # 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}` in our mailbox — that's what tells a
    # standby client the leader vanished and election should
    # re-run.
    :ok = :net_kernel.monitor_nodes(true)

    state = %{
      role: :standby,
      socket: nil,
      buffer: <<>>,
      bands: Keyword.get(opts, :bands, @default_bands),
      aggregator: Keyword.get(opts, :aggregator, Aggregator),
      client_id: Keyword.get(opts, :client_id, default_client_id()),
      received: 0,
      ping_timer: nil
    }

    {: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 ->
        Logger.info("Pskr.Client standby on #{node()} (leader: #{inspect(:global.whereis_name(__MODULE__))})")
        {:noreply, state}
    end
  end

  def handle_continue(:connect, state) do
    case do_connect(state) do
      {:ok, socket, ping_timer} ->
        {:noreply, %{state | socket: socket, ping_timer: ping_timer}}

      {: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}}

  def handle_info(:ping, %{socket: socket} = state) when is_port(socket) do
    case :gen_tcp.send(socket, Mqtt.pingreq()) do
      :ok ->
        {:noreply, %{state | ping_timer: schedule_ping()}}

      {:error, reason} ->
        Logger.warning("Pskr.Client PINGREQ send failed: #{inspect(reason)}")
        {:noreply, drop_socket(state), {:continue, :connect}}
    end
  end

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

  # Standby clients re-run election on any cluster topology change.
  # Active leaders ignore — their registration stays valid until
  # *their* node disappears.
  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}

  def handle_info({:tcp, socket, data}, %{socket: socket} = state) do
    :inet.setopts(socket, active: :once)
    {:noreply, process_buffer(%{state | buffer: state.buffer <> data})}
  end

  def handle_info({:tcp_closed, socket}, %{socket: socket} = state) do
    Logger.warning("Pskr.Client TCP closed (received #{state.received} so far) — reconnecting in 5s")
    Process.send_after(self(), :reconnect, 5_000)
    {:noreply, drop_socket(state)}
  end

  def handle_info({:tcp_error, socket, reason}, %{socket: socket} = state) do
    Logger.warning("Pskr.Client TCP error #{inspect(reason)} — reconnecting in 5s")
    Process.send_after(self(), :reconnect, 5_000)
    {:noreply, drop_socket(state)}
  end

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

  @impl true
  def terminate(_reason, %{socket: socket}) when is_port(socket) do
    _ = :gen_tcp.send(socket, Mqtt.disconnect())
    _ = :gen_tcp.close(socket)
    :ok
  end

  def terminate(_reason, _state), do: :ok

  # ---------- Connection setup ----------

  defp do_connect(state) do
    tcp_opts = [:binary, active: :once, packet: :raw, keepalive: true]

    with {:ok, socket} <- :gen_tcp.connect(@broker_host, @broker_port, tcp_opts, 10_000),
         :ok <- :gen_tcp.send(socket, Mqtt.connect(state.client_id, @keepalive_seconds)),
         {:ok, :connected, rest} <- await_connack(socket, <<>>),
         :ok <- :gen_tcp.send(socket, subscribe_packet(state.bands)) do
      Logger.info("Pskr.Client connected as #{state.client_id}, subscribed to #{length(state.bands)} bands")
      # `rest` holds anything the broker sent piggybacked after
      # CONNACK — feed it back into the buffer so we don't lose
      # the SUBACK or an early PUBLISH.
      send(self(), {:tcp, socket, rest})
      {:ok, socket, schedule_ping()}
    end
  end

  # CONNACK arrives synchronously right after CONNECT, so a small
  # blocking read here keeps the state machine simple. If we don't
  # get it within 10s, treat the connection as dead.
  defp await_connack(socket, buffer) do
    case Mqtt.parse(buffer) do
      {:ok, {:connack, 0, _}, rest} ->
        {:ok, :connected, rest}

      {:ok, {:connack, code, _}, _} ->
        {:error, {:connack_refused, code}}

      {:incomplete, buf} ->
        case :gen_tcp.recv(socket, 0, 10_000) do
          {:ok, chunk} -> await_connack(socket, buf <> chunk)
          {:error, _} = err -> err
        end

      {:error, _} = err ->
        err
    end
  end

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

    Mqtt.subscribe(1, topics)
  end

  # Topic order (from the broker's landing page):
  #   pskr/filter/v2/<band>/<mode>/<sc>/<rc>/<sl>/<rl>/<sa>/<ra>
  defp topic_for(band, :sender_us), do: "pskr/filter/v2/#{band}/+/+/+/+/+/#{@us_dxcc}/+"
  defp topic_for(band, :receiver_us), do: "pskr/filter/v2/#{band}/+/+/+/+/+/+/#{@us_dxcc}"

  # ---------- Inbound dispatch ----------

  defp process_buffer(state) do
    case Mqtt.parse(state.buffer) do
      {:ok, {:publish, _topic, payload}, rest} ->
        Aggregator.ingest(state.aggregator, payload)
        process_buffer(%{state | buffer: rest, received: state.received + 1})

      {:ok, {:suback, _id, results}, rest} ->
        if Enum.any?(results, &(&1 == 0x80)) do
          Logger.warning("Pskr.Client SUBACK contains failures: #{inspect(results)}")
        end

        process_buffer(%{state | buffer: rest})

      {:ok, :pingresp, rest} ->
        process_buffer(%{state | buffer: rest})

      {:ok, _other, rest} ->
        process_buffer(%{state | buffer: rest})

      {:incomplete, buffer} ->
        %{state | buffer: buffer}

      {:error, reason} ->
        Logger.error("Pskr.Client parse error: #{inspect(reason)} — dropping connection")
        Process.send_after(self(), :reconnect, 1_000)
        drop_socket(state)
    end
  end

  defp drop_socket(state) do
    if state.ping_timer, do: Process.cancel_timer(state.ping_timer)
    if is_port(state.socket), do: :gen_tcp.close(state.socket)
    %{state | socket: nil, buffer: <<>>, ping_timer: nil}
  end

  defp schedule_ping, do: Process.send_after(self(), :ping, @keepalive_ms)

  # ---------- Client ID ----------

  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