OpsLog/internal/pskr/pskr.go

// Package pskr subscribes to PSK Reporter's MQTT feed and turns it into the
// spots the band-opening detector already eats.
//
// Why this exists at all: the detector was fed from the DX cluster and the RBN,
// and on VHF that is a few hundred skimmers, nearly all of them on HF. A 6 m
// opening carrying 869 stations reached OpsLog as a handful of spots, or none.
// PSK Reporter is every ordinary station running WSJT-X and reporting what it
// decodes — the difference is two orders of magnitude, not a threshold.
//
// The feed's shape happens to suit us exactly:
//
//	topic   pskr/filter/v2/<band>/<mode>/<tx call>/<rx call>/...
//	payload {"f":50313000,"md":"FT8","rp":-12,"sc":"F4BPO","sl":"JN36",
//	         "rc":"OH5CX","rl":"KP30","b":"6m"}
//
// BOTH grids are in the message, so distance and bearing are arithmetic. No
// lookup, no DXCC-centre approximation, no extra network call — which is what
// made the cluster path's bearings coarse.
//
// Volume is the real design constraint. Six metres open is thousands of
// messages a minute, and OpsLog runs on some very old PCs. So: no history is
// kept here, nothing is persisted, each message is parsed and handed on or
// dropped, and the callback does the deciding.
package pskr

import (
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

// DefaultBroker is PSK Reporter's public MQTT endpoint, TLS.
const DefaultBroker = "tls://mqtt.pskreporter.info:1884"

// DefaultNearKm is the radius that counts as "around here" for a receiver.
const DefaultNearKm = 300

// Bands offered. Anything below 10 m is deliberately absent, 12 m included: an
// "opening" on 20 m is the normal state of the band and announcing it says
// nothing, and 12 m is close enough to that at this point in the cycle to be the
// same problem. These are the bands where an opening is an event.
//
// Fewer subscriptions is also less traffic, which matters here: the operator's
// PC pays for every message the broker sends, whether or not anything comes of it.
var Bands = []string{"10m", "6m", "4m", "2m"}

// Spot is one decode, already reduced to what a detector needs.
type Spot struct {
	Call    string // the transmitting station
	Band    string
	Mode    string
	Grid    string // transmitter's grid, 4 characters
	DistKm  int    // from the operator
	Bearing int    // degrees from the operator, short path
	At      time.Time
}

// Config is what the watcher needs to run.
type Config struct {
	Broker string
	Bands  []string
	// OpLat/OpLon are the operator's position: every spot is measured from it,
	// so with no position there is nothing to measure and the watcher stays down.
	OpLat, OpLon float64
	// NearKm is how close a RECEIVER must be to count as "around here". A report
	// collected further away proves a path that is not the operator's.
	//
	// 300 km by default: far enough to borrow the ears of a whole region, which
	// is the point — an opening reaches an area, not a postcode, and waiting for
	// a decode at one's own station is just working the band. Close enough that
	// the ionosphere doing something there is the ionosphere doing it here.
	NearKm int
	// Geo turns two grids into distance and bearing. Injected rather than
	// implemented here so it stays the SAME arithmetic the cluster path uses —
	// two answers for one question is how a bearing quietly becomes wrong.
	Geo func(grid string) (distKm int, bearing int, ok bool)
	// OnSpot receives every accepted decode. Called from the MQTT goroutine, so
	// it must not block: the broker's buffer is what pays for it if it does.
	OnSpot func(Spot)
	// OnGrid receives the transmitter of EVERY message, before any geographic
	// filtering, for the callsign-to-locator store. Same goroutine as OnSpot and
	// the same rule: do not block.
	OnGrid func(call, grid string)
	// RxGrids filters at the BROKER: only reports collected by a receiver in one
	// of these squares are sent at all. Empty keeps the old behaviour, which was
	// to receive the world and discard it here — measured at 83 messages a second
	// for the four opening bands, of which about one in a hundred survived.
	RxGrids []string
	Logf    func(string, ...any)
}

// Watcher owns the MQTT connection and its subscriptions.
type Watcher struct {
	mu      sync.Mutex
	cfg     Config
	client  mqtt.Client
	running bool

	// received counts accepted spots since start, for the status panel: a
	// connection that is up but silent looks identical to one that is working
	// until you can see a number moving.
	received uint64
	lastAt   time.Time
	lastErr  string
}

func New(cfg Config) *Watcher {
	if cfg.Broker == "" {
		cfg.Broker = DefaultBroker
	}
	if len(cfg.Bands) == 0 {
		cfg.Bands = Bands
	}
	if cfg.NearKm <= 0 {
		cfg.NearKm = DefaultNearKm
	}
	if cfg.Logf == nil {
		cfg.Logf = func(string, ...any) {}
	}
	return &Watcher{cfg: cfg}
}

// topics builds the subscription list.
//
// The v2 topic is
//
//	pskr/filter/v2/<band>/<mode>/<tx call>/<rx call>/<tx grid>/<rx grid>/<tx dxcc>/<rx dxcc>
//
// so the receiver's square is a level the broker can filter on, and a band of
// "+" means every band. Filtering by RECEIVER square rather than by receiver
// DXCC is deliberate: measured on 20 m, one country ranged from 1.2 messages a
// second (OH) to 72.5 (K), because a DXCC can be a continent. By square the
// same measurement is 0.2 to 1.2 — the load follows distance, which is what the
// feed is actually about, and it is the same for every operator.
func (w *Watcher) topics() []string {
	out := []string{}
	for _, b := range w.cfg.Bands {
		if len(w.cfg.RxGrids) == 0 {
			out = append(out, "pskr/filter/v2/"+b+"/#")
			continue
		}
		for _, g := range w.cfg.RxGrids {
			out = append(out, "pskr/filter/v2/"+b+"/+/+/+/+/"+strings.ToUpper(g)+"/+/+")
		}
	}
	return out
}

// Start connects and subscribes. Safe to call when already running.
func (w *Watcher) Start() error {
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.running {
		return nil
	}
	if w.cfg.Geo == nil || w.cfg.OnSpot == nil {
		return fmt.Errorf("pskr: Geo and OnSpot are required")
	}

	opts := mqtt.NewClientOptions().
		AddBroker(w.cfg.Broker).
		// A stable client id would collide with another OpsLog on the same
		// account; the broker is anonymous, so uniqueness is ours to provide.
		SetClientID(fmt.Sprintf("opslog-%d", time.Now().UnixNano())).
		SetCleanSession(true).
		SetAutoReconnect(true).
		SetConnectRetry(true).
		SetConnectRetryInterval(30 * time.Second).
		SetConnectTimeout(15 * time.Second).
		// No message is worth keeping if we cannot handle it now: an opening is
		// a thing happening at this moment, and a queue of stale decodes would
		// announce one that finished twenty minutes ago.
		SetOrderMatters(false)

	opts.OnConnect = func(c mqtt.Client) {
		w.cfg.Logf("pskr: connected to %s", w.cfg.Broker)
		for _, topic := range w.topics() {
			if tok := c.Subscribe(topic, 0, w.handle); tok.Wait() && tok.Error() != nil {
				w.cfg.Logf("pskr: subscribe %s failed: %v", topic, tok.Error())
				continue
			}
			w.cfg.Logf("pskr: watching %s", topic)
		}
	}
	opts.OnConnectionLost = func(_ mqtt.Client, err error) {
		w.mu.Lock()
		w.lastErr = err.Error()
		w.mu.Unlock()
		w.cfg.Logf("pskr: connection lost: %v (will retry)", err)
	}

	c := mqtt.NewClient(opts)
	// Deliberately NOT waiting on the connect token: the broker may be slow or
	// unreachable and startup must not hang on a feature that is decoration.
	// ConnectRetry brings it up in the background when it can.
	c.Connect()
	w.client = c
	w.running = true
	return nil
}

// Stop disconnects. Safe to call when already stopped.
func (w *Watcher) Stop() {
	w.mu.Lock()
	c, running := w.client, w.running
	w.client, w.running = nil, false
	w.mu.Unlock()
	if running && c != nil {
		c.Disconnect(250)
	}
}

// wire is the payload as PSK Reporter sends it — short keys, no nesting.
type wire struct {
	Freq   int64  `json:"f"`
	Mode   string `json:"md"`
	SNR    int    `json:"rp"`
	TxCall string `json:"sc"`
	TxGrid string `json:"sl"`
	RxCall string `json:"rc"`
	RxGrid string `json:"rl"`
	Band   string `json:"b"`
}

func (w *Watcher) handle(_ mqtt.Client, m mqtt.Message) {
	var p wire
	if err := json.Unmarshal(m.Payload(), &p); err != nil {
		return // malformed payloads are not worth a log line at this rate
	}
	call := strings.ToUpper(strings.TrimSpace(p.TxCall))
	grid := strings.ToUpper(strings.TrimSpace(p.TxGrid))
	rxGrid := strings.ToUpper(strings.TrimSpace(p.RxGrid))
	if call == "" || len(grid) < 4 || len(rxGrid) < 4 {
		return
	}

	// The locator store takes every transmitter, before any of the geography
	// below. What it wants is "which square is this callsign in", and that is
	// true whoever happened to hear the report.
	if w.cfg.OnGrid != nil {
		w.cfg.OnGrid(call, grid[:4])
	}

	// THE RECEIVER HAS TO BE NEAR THE OPERATOR. This is the whole difference
	// between a useful feed and a world map.
	//
	// A PSK Reporter message says "X was heard BY Y". Without this check, a
	// station 1400 km from here heard by somebody in Japan counted as evidence
	// of an opening — it proves the path from X to JAPAN, and says nothing at all
	// about whether anything reaches this station. That is how a "2 m opening"
	// came to be announced from a KX9X in the United States.
	//
	// Only reports collected by a receiver in the operator's own region show that
	// signals are actually arriving HERE, which is the only question worth asking.
	if rxDist, _, ok := w.cfg.Geo(rxGrid[:4]); !ok || rxDist > w.cfg.NearKm {
		return
	}

	// The TRANSMITTER is the station on the air, so it is the transmitter's grid
	// that gives the direction and length of the path that just proved itself.
	dist, brg, ok := w.cfg.Geo(grid[:4])
	if !ok {
		return
	}
	s := Spot{
		Call: call, Band: strings.ToLower(strings.TrimSpace(p.Band)),
		Mode: strings.ToUpper(strings.TrimSpace(p.Mode)), Grid: grid[:4],
		DistKm: dist, Bearing: brg,
		// Stamped on receipt: the broker's own timestamps vary between payload
		// versions, and the window this feeds is measured in minutes.
		At: time.Now(),
	}
	w.mu.Lock()
	w.received++
	w.lastAt = s.At
	w.mu.Unlock()
	w.cfg.OnSpot(s)
}

// Status is the snapshot the settings panel shows.
type Status struct {
	Running  bool      `json:"running"`
	Received uint64    `json:"received"`
	LastAt   time.Time `json:"last_at"`
	LastErr  string    `json:"last_err,omitempty"`
	Broker   string    `json:"broker"`
	Bands    []string  `json:"bands"`
}

func (w *Watcher) Status() Status {
	w.mu.Lock()
	defer w.mu.Unlock()
	st := Status{
		Running: w.running, Received: w.received, LastAt: w.lastAt,
		LastErr: w.lastErr, Broker: w.cfg.Broker,
	}
	st.Bands = append(st.Bands, w.cfg.Bands...)
	return st
}