OpsLog/internal/pskrtgt/pskrtgt.go

// Package pskrtgt answers one question about one station: can they hear me?
//
// It is the other way round from internal/pskr. That watcher asks what is
// happening AROUND HERE — reports collected near the operator, whoever sent
// them — and it is the right shape for finding a band opening or a new entity.
// This one starts from a callsign the operator wants to work and gathers the
// evidence about that path, in both directions:
//
//   - did the DX decode MY call, and how long ago
//   - who NEAR ME did the DX decode (the path is open at my end)
//   - who near the DX decoded ME (the path is open at his end, even when he
//     uploads nothing himself)
//   - how many stations he is decoding right now (the pileup I am up against)
//   - where in his receive passband those decodes land, so a caller can pick a
//     slot he is not already covered on
//
// Nothing here is persisted and nothing is inferred from a QSO: it is a sliding
// window of PSK Reporter reports, and when the window empties the answer goes
// back to "not known", which is the honest answer.
//
// The window is FIVE minutes. An FT8 cycle is fifteen seconds, so that is
// twenty chances for a path to show itself — short enough that "he decoded you"
// still means now, long enough that one missed cycle does not erase it.
package pskrtgt

import (
	"encoding/json"
	"encoding/xml"
	"fmt"
	"net/http"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

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

// DefaultBroker is PSK Reporter's public MQTT endpoint, TLS. Same one the
// band-opening watcher uses — two connections to it, because the two want
// opposite slices of the feed and neither can be filtered out of the other's.
const DefaultBroker = "tls://mqtt.pskreporter.info:1884"

const (
	// window is how far back a report still counts.
	//
	// TEN minutes. Five was chosen as "recent enough to still mean now", and on
	// the air it meant half the evidence: PSK Reporter's uploaders batch their
	// reports, many of them every five minutes, so a five-minute window catches
	// roughly one upload cycle per station. Side by side with DXHunter on the
	// same DX, the same second: 18 decodes here against 27 there, and a station
	// missing from "from your area" that was simply six minutes old.
	//
	// It is a window on ONE station's activity, not on the band: ten minutes of
	// a DX working a pileup is still what he is doing now.
	window = 10 * time.Minute
	// pileupWindow is the tighter one for "how many stations is he working
	// through". A station he decoded four minutes ago has very likely moved on,
	// and counting it inflates the only number an operator uses to decide
	// whether it is worth calling at all.
	pileupWindow = 2 * time.Minute

	// backfillEvery is how often the history query is asked again for a target
	// that is still being watched. Five minutes is PSK Reporter's own courtesy
	// interval for repeating a query, and it happens to be the period most
	// uploaders batch on.
	backfillEvery = 5 * time.Minute

	// The passband histogram: 60 Hz bins from 200 Hz to 4000 Hz. Above 4 kHz
	// there is essentially no FT8, and drawing the empty space made the strip
	// look broken rather than empty.
	binHz  = 60
	lowHz  = 200
	highHz = 4000
)

// Scope decides how much of the feed is subscribed to.
type Scope string

const (
	// ScopeTarget subscribes to three filters: what the DX transmits, what he
	// receives, and who hears the operator. A handful of messages a second, and
	// the REST backfill fills the window the moment the target changes.
	ScopeTarget Scope = "target"
	// ScopeBand subscribes to the whole band's FTx traffic. Switching target is
	// then instant with no backfill, at the cost of every message on the band —
	// hundreds a second when 20 m is busy.
	ScopeBand Scope = "band"
)

// spot is one PSK Reporter reception report, as the v2 payload carries it.
type spot 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"`
	// at is stamped on arrival. The payload's own timestamps differ between
	// versions of the feed, and everything here is measured in minutes.
	at time.Time
}

// Entry is one station in one of the lists the panel shows.
type Entry struct {
	Call     string `json:"call"`
	Grid     string `json:"grid"`
	SNR      int    `json:"snr"`
	OffsetHz int    `json:"offset_hz"` // audio offset from the operator's dial
	AgeSec   int    `json:"age_sec"`
}

// Bin is one 60 Hz slice of the DX's receive passband.
type Bin struct {
	OffsetHz int     `json:"offset_hz"`
	Count    int     `json:"count"`
	AvgSNR   float64 `json:"avg_snr"`
}

// Analysis is the whole snapshot the panel draws, recomputed on demand.
type Analysis struct {
	Target string `json:"target"`
	Mode   string `json:"mode,omitempty"`
	// Enabled is the operator's switch; Online is whether the broker is
	// actually connected. A panel that says nothing has to be able to say WHY.
	Enabled bool `json:"enabled"`
	Online  bool `json:"online"`
	// Spots is everything in the window, the sign that the feed is alive even
	// when every counter below is legitimately zero.
	Spots int `json:"spots"`

	// HeMe is the answer to the question. The rest is what to do when it is no.
	HeMe        bool `json:"he_me"`
	HeMeSeconds int  `json:"he_me_seconds"`
	HeMeSNR     int  `json:"he_me_snr"`
	HeMeOffset  int  `json:"he_me_offset_hz"`

	// TargetUploads distinguishes "he is not hearing anybody" from "his software
	// tells PSK Reporter nothing" — without it, a silent panel reads as a dead
	// band when it may be a full one.
	TargetUploads bool   `json:"target_uploads"`
	TargetGrid    string `json:"target_grid,omitempty"`

	// Near the DX: stations in his square that decoded the operator. This is
	// what still works when he uploads nothing himself.
	NearHimCount int     `json:"near_him_count"`
	NearHimTop   []Entry `json:"near_him_top"`

	// Near the operator: stations in his own field that the DX decoded.
	FromMyAreaCount int     `json:"from_my_area_count"`
	FromMyAreaTop   []Entry `json:"from_my_area_top"`
	PathOpen        bool    `json:"path_open"`

	// Who heard the DX, worldwide and locally.
	HeardByCount   int     `json:"heard_by_count"`
	HeardNearMe    int     `json:"heard_near_me"`
	HeardNearMeTop []Entry `json:"heard_near_me_top"`

	// The pileup: everyone he decoded (window), and the recent slice of it.
	DecodedByCount int      `json:"decoded_by_count"`
	DecodedByTop   []Entry  `json:"decoded_by_top"`
	DecodedByCalls []string `json:"decoded_by_calls"`
	PileupCount    int      `json:"pileup_count"`

	// His receive passband, and a slot in it that nobody is using.
	DialHz          int64 `json:"dial_hz"`
	CeilingHz       int   `json:"ceiling_hz"`
	DecodesInWindow int   `json:"decodes_in_window"`
	Bins            []Bin `json:"bins"`
	SuggestedOffset int   `json:"suggested_offset"`
}

// Config is what the watcher needs from the application.
type Config struct {
	Broker string
	Scope  Scope
	// MyCall and MyGrid are the operator's. Both matter: the callsign is what
	// "he decoded you" is looked up by, and the grid decides what counts as
	// "near me" — its first two characters, a Maidenhead FIELD, which is a few
	// hundred kilometres rather than a whole continent.
	MyCall string
	MyGrid string
	// Continent resolves a callsign to EU/NA/AS/… It is only a FALLBACK, for an
	// operator whose grid is not set: without a grid there is nothing to compare
	// squares with, and a continent is better than nothing. Injected so this
	// package does not pull in the country file.
	Continent func(call string) string
	Logf      func(string, ...any)
}

// Watcher owns the MQTT connection and the window.
type Watcher struct {
	mu     sync.Mutex
	cfg    Config
	client mqtt.Client

	target string // the callsign being analysed, upper case
	mode   string // FT8 / FT4 — the target's mode, for the band-scope topic
	band   string // band tag currently subscribed to under ScopeBand
	dialHz int64  // the operator's dial, for audio offsets

	// subs is what we are subscribed to right now, so a target change can take
	// the old filters down without guessing at their shape.
	subs  []string
	spots []spot

	// backfilled remembers the target the REST history was fetched for and when,
	// so the panel's polling cannot re-fetch it every second — PSK Reporter's
	// query API answers that with a rate limit, and rightly — while a target
	// held for a while still gets a fresh look every few minutes.
	backfilled   string
	backfilledAt time.Time
}

func New(cfg Config) *Watcher {
	if cfg.Broker == "" {
		cfg.Broker = DefaultBroker
	}
	if cfg.Scope == "" {
		cfg.Scope = ScopeTarget
	}
	if cfg.Logf == nil {
		cfg.Logf = func(string, ...any) {}
	}
	return &Watcher{cfg: cfg}
}

// Watch points the analysis at a callsign. Connects on the first call, so an
// operator who never opens the panel never opens a socket.
//
// Called repeatedly with the same target — the panel re-asserts it as the
// operator works — so everything expensive here is guarded on an actual change.
func (w *Watcher) Watch(target, mode string, dialHz int64) error {
	target = strings.ToUpper(strings.TrimSpace(target))
	mode = strings.ToUpper(strings.TrimSpace(mode))
	if mode == "" {
		mode = "FT8"
	}
	if target == "" {
		w.Stop()
		return nil
	}

	w.mu.Lock()
	changed := target != w.target || mode != w.mode
	w.target, w.mode = target, mode
	if dialHz > 0 {
		w.dialHz = dialHz
	}
	band := bandTag(w.dialHz)
	bandChanged := band != "" && band != w.band
	// Set BEFORE any connect: the subscription is built from it, and a first
	// connect that found it empty would subscribe to every band at once under
	// the band-wide scope — the one case where that is expensive.
	if band != "" {
		w.band = band
	}
	client := w.client
	w.mu.Unlock()

	if client == nil || !client.IsConnected() {
		c, err := w.connect()
		if err != nil {
			return err
		}
		w.mu.Lock()
		w.client, client = c, c
		w.mu.Unlock()
		// connect() subscribes on its own OnConnect handler; anything below
		// would only repeat it.
		changed = false
		bandChanged = false
	}
	if !changed && !bandChanged {
		return nil
	}

	w.mu.Lock()
	// The window belongs to the target it was collected for. Keeping it across a
	// change would answer the new question with the old station's evidence.
	if changed {
		w.spots = w.spots[:0]
	}
	w.mu.Unlock()

	if err := w.resubscribe(client); err != nil {
		return err
	}
	if changed {
		// In BOTH scopes. The band-wide subscription was assumed to arrive with
		// the target's reports already in the window — true only once it has been
		// running a while, and false in the case that matters: the operator picks
		// a station a minute after opening the panel and sees one decode where
		// another program, running for an hour, shows four.
		go w.backfill(target, mode)
	}
	return nil
}

// resubscribe replaces every filter with the ones the current target and scope
// want. Takes the old ones down first: a target change that only ADDED filters
// would leave the previous station's reports arriving for ever.
func (w *Watcher) resubscribe(c mqtt.Client) error {
	w.mu.Lock()
	old := w.subs
	topics := w.topicsLocked()
	w.subs = topics
	target, scope := w.target, w.cfg.Scope
	w.mu.Unlock()

	if len(old) > 0 {
		if tok := c.Unsubscribe(old...); tok.Wait() && tok.Error() != nil {
			w.cfg.Logf("pskr target: unsubscribe failed: %v", tok.Error())
		}
	}
	if len(topics) == 0 {
		return nil
	}
	filters := make(map[string]byte, len(topics))
	for _, t := range topics {
		filters[t] = 0
	}
	if tok := c.SubscribeMultiple(filters, w.handle); tok.Wait() && tok.Error() != nil {
		return fmt.Errorf("pskr target: subscribe: %w", tok.Error())
	}
	w.cfg.Logf("pskr target: watching %s (%s scope, %d filters)", target, scope, len(topics))
	return nil
}

// topicsLocked 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 both directions of one callsign are addressable at the broker, which is
// the whole reason the narrow scope costs almost nothing.
func (w *Watcher) topicsLocked() []string {
	if w.target == "" {
		return nil
	}
	if w.cfg.Scope == ScopeBand {
		band := w.band
		if band == "" {
			band = "+"
		}
		return []string{"pskr/filter/v2/" + band + "/" + w.mode + "/#"}
	}
	out := []string{
		// What he is transmitting: who is hearing him.
		"pskr/filter/v2/+/" + w.mode + "/" + w.target + "/#",
		// What he is receiving: the pileup, and whether the operator is in it.
		"pskr/filter/v2/+/" + w.mode + "/+/" + w.target + "/#",
	}
	// Who hears the OPERATOR. Only some of those receivers are near the DX, and
	// those are the ones that answer "can I be heard over there" on a DX who
	// uploads nothing himself. Left out when the callsign is not configured
	// rather than subscribing to a filter with an empty level in it.
	if my := strings.ToUpper(strings.TrimSpace(w.cfg.MyCall)); my != "" {
		out = append(out, "pskr/filter/v2/+/"+w.mode+"/"+my+"/#")
	}
	return out
}

func (w *Watcher) connect() (mqtt.Client, error) {
	opts := mqtt.NewClientOptions().
		AddBroker(w.cfg.Broker).
		SetClientID(fmt.Sprintf("opslog-tgt-%d", time.Now().UnixNano())).
		SetCleanSession(true).
		SetAutoReconnect(true).
		SetConnectRetry(true).
		SetConnectRetryInterval(30 * time.Second).
		SetConnectTimeout(15 * time.Second).
		SetOrderMatters(false)
	// Re-subscribe on every connect, reconnects included: the session is clean,
	// so the broker remembers nothing and a dropped link would otherwise come
	// back up subscribed to nothing at all — a panel that goes quiet for ever
	// while still saying "online".
	opts.OnConnect = func(c mqtt.Client) {
		w.mu.Lock()
		w.subs = nil
		target, mode := w.target, w.mode
		w.mu.Unlock()
		if err := w.resubscribe(c); err != nil {
			w.cfg.Logf("pskr target: %v", err)
		}
		if target != "" {
			go w.backfill(target, mode)
		}
	}
	opts.OnConnectionLost = func(_ mqtt.Client, err error) {
		w.cfg.Logf("pskr target: connection lost: %v (will retry)", err)
	}
	c := mqtt.NewClient(opts)
	tok := c.Connect()
	if !tok.WaitTimeout(15*time.Second) || tok.Error() != nil {
		err := tok.Error()
		if err == nil {
			err = fmt.Errorf("timeout")
		}
		return nil, fmt.Errorf("pskr target: connect %s: %w", w.cfg.Broker, err)
	}
	return c, nil
}

func (w *Watcher) handle(_ mqtt.Client, m mqtt.Message) {
	var s spot
	if err := json.Unmarshal(m.Payload(), &s); err != nil {
		return
	}
	if s.TxCall == "" || s.RxCall == "" {
		return
	}
	s.TxCall = strings.ToUpper(s.TxCall)
	s.RxCall = strings.ToUpper(s.RxCall)
	s.TxGrid = strings.ToUpper(s.TxGrid)
	s.RxGrid = strings.ToUpper(s.RxGrid)
	s.at = time.Now()
	w.mu.Lock()
	w.spots = append(w.spots, s)
	w.mu.Unlock()
}

// Stop drops the target and the connection. The window goes with it: it is
// evidence about a station nobody is asking about any more.
func (w *Watcher) Stop() {
	w.mu.Lock()
	c := w.client
	w.client, w.target, w.band, w.subs, w.backfilled = nil, "", "", nil, ""
	w.spots = nil
	w.mu.Unlock()
	if c != nil {
		c.Disconnect(250)
	}
}

// SetDial updates the frequency audio offsets are measured against.
func (w *Watcher) SetDial(hz int64) {
	if hz <= 0 {
		return
	}
	w.mu.Lock()
	w.dialHz = hz
	w.mu.Unlock()
}

// SetOperator refreshes the operator's own callsign and grid. Called when the
// station profile changes: every "near me" answer is measured from these, and a
// stale pair would quietly measure them from somebody else's station.
func (w *Watcher) SetOperator(call, grid string) {
	w.mu.Lock()
	w.cfg.MyCall = strings.ToUpper(strings.TrimSpace(call))
	w.cfg.MyGrid = strings.ToUpper(strings.TrimSpace(grid))
	w.mu.Unlock()
}

// Snapshot recomputes the analysis from the window.
func (w *Watcher) Snapshot() Analysis {
	// Due another look at the history? Checked here because this is what the
	// panel calls every second; the query itself is rate-limited inside
	// backfill, so this cannot turn into a request per poll.
	w.mu.Lock()
	if t, m := w.target, w.mode; t != "" && time.Since(w.backfilledAt) >= backfillEvery {
		w.mu.Unlock()
		go w.backfill(t, m)
		w.mu.Lock()
	}
	defer w.mu.Unlock()

	now := time.Now()
	cutoff := now.Add(-window)
	kept := w.spots[:0]
	for _, s := range w.spots {
		if s.at.After(cutoff) {
			kept = append(kept, s)
		}
	}
	w.spots = kept

	a := Analysis{
		Target: w.target,
		Mode:   w.mode,
		Online: w.client != nil && w.client.IsConnected(),
		DialHz: w.dialHz,
		Spots:  len(w.spots),
	}
	if w.target == "" {
		return a
	}

	myCall := strings.ToUpper(strings.TrimSpace(w.cfg.MyCall))
	myField := ""
	if g := strings.ToUpper(strings.TrimSpace(w.cfg.MyGrid)); len(g) >= 2 {
		myField = g[:2]
	}
	myCont := ""
	if myField == "" && myCall != "" && w.cfg.Continent != nil {
		myCont = strings.ToUpper(w.cfg.Continent(myCall))
	}

	// His square, taken from any report where he was transmitting. It is what
	// "near him" is measured against, so without it that whole answer is
	// unavailable rather than approximated.
	for i := range w.spots {
		if w.spots[i].TxCall == w.target && len(w.spots[i].TxGrid) >= 4 {
			a.TargetGrid = w.spots[i].TxGrid[:4]
		}
	}

	entry := func(call, grid string, s *spot) Entry {
		off := 0
		if w.dialHz > 0 {
			off = int(s.Freq - w.dialHz)
		}
		return Entry{Call: call, Grid: grid, SNR: s.SNR, OffsetHz: off,
			AgeSec: int(now.Sub(s.at).Seconds())}
	}
	// One entry per station, overwritten as newer reports arrive, so a station
	// calling every cycle counts once and shows its latest report.
	heardBy := map[string]Entry{}
	heardNearMe := map[string]Entry{}
	fromMyArea := map[string]Entry{}
	decodedBy := map[string]Entry{}
	nearHim := map[string]Entry{}
	pileup := map[string]struct{}{}
	pileupCutoff := now.Add(-pileupWindow)

	type acc struct {
		n   int
		sum float64
	}
	bins := map[int]*acc{}
	var lastHeMe *spot

	near := func(theirGrid, call string) bool {
		if myField != "" {
			return strings.HasPrefix(strings.ToUpper(theirGrid), myField)
		}
		if myCont != "" && w.cfg.Continent != nil {
			return strings.ToUpper(w.cfg.Continent(call)) == myCont
		}
		return false
	}

	for i := range w.spots {
		s := &w.spots[i]

		// He transmitted: somebody heard him.
		if s.TxCall == w.target {
			e := entry(s.RxCall, s.RxGrid, s)
			heardBy[s.RxCall] = e
			if near(s.RxGrid, s.RxCall) {
				heardNearMe[s.RxCall] = e
			}
		}

		// The operator transmitted and a station in the DX's own square heard
		// it. That is a path to his region, proved without his help.
		if a.TargetGrid != "" && myCall != "" && s.TxCall == myCall &&
			strings.HasPrefix(s.RxGrid, a.TargetGrid) {
			nearHim[s.RxCall] = entry(s.RxCall, s.RxGrid, s)
		}

		// He received: this is the pileup, the passband, and the answer.
		if s.RxCall == w.target {
			a.DecodesInWindow++
			if s.TxCall == myCall {
				if lastHeMe == nil || s.at.After(lastHeMe.at) {
					lastHeMe = s
				}
				continue // the operator is not part of his own pileup
			}
			decodedBy[s.TxCall] = entry(s.TxCall, s.TxGrid, s)
			if s.at.After(pileupCutoff) {
				pileup[s.TxCall] = struct{}{}
			}
			if near(s.TxGrid, s.TxCall) {
				fromMyArea[s.TxCall] = entry(s.TxCall, s.TxGrid, s)
			}
			if w.dialHz > 0 {
				off := int(s.Freq - w.dialHz)
				if off >= lowHz && off <= highHz {
					edge := (off / binHz) * binHz
					b := bins[edge]
					if b == nil {
						b = &acc{}
						bins[edge] = b
					}
					b.n++
					b.sum += float64(s.SNR)
					if off > a.CeilingHz {
						a.CeilingHz = off
					}
				}
			}
		}
	}

	if lastHeMe != nil {
		a.HeMe = true
		a.HeMeSeconds = int(now.Sub(lastHeMe.at).Seconds())
		a.HeMeSNR = lastHeMe.SNR
		if w.dialHz > 0 {
			a.HeMeOffset = int(lastHeMe.Freq - w.dialHz)
		}
	}
	a.TargetUploads = a.DecodesInWindow > 0
	a.HeardByCount = len(heardBy)
	a.HeardNearMe = len(heardNearMe)
	a.HeardNearMeTop = top(heardNearMe, 5)
	a.FromMyAreaCount = len(fromMyArea)
	a.FromMyAreaTop = top(fromMyArea, 5)
	a.PathOpen = a.FromMyAreaCount > 0
	a.NearHimCount = len(nearHim)
	a.NearHimTop = top(nearHim, 5)
	a.DecodedByCount = len(decodedBy)
	a.DecodedByTop = top(decodedBy, 10)
	a.DecodedByCalls = make([]string, 0, len(decodedBy))
	for c := range decodedBy {
		a.DecodedByCalls = append(a.DecodedByCalls, c)
	}
	sort.Strings(a.DecodedByCalls)
	a.PileupCount = len(pileup)

	a.Bins = make([]Bin, 0, len(bins))
	for edge, b := range bins {
		avg := 0.0
		if b.n > 0 {
			avg = b.sum / float64(b.n)
		}
		a.Bins = append(a.Bins, Bin{OffsetHz: edge, Count: b.n, AvgSNR: avg})
	}
	sort.Slice(a.Bins, func(i, j int) bool { return a.Bins[i].OffsetHz < a.Bins[j].OffsetHz })
	a.SuggestedOffset = suggestOffset(a.Bins, a.CeilingHz)
	return a
}

// top returns the freshest entries from a per-callsign map, newest first.
func top(m map[string]Entry, limit int) []Entry {
	out := make([]Entry, 0, len(m))
	for _, e := range m {
		out = append(out, e)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].AgeSec < out[j].AgeSec })
	if len(out) > limit {
		out = out[:limit]
	}
	return out
}

// suggestOffset picks an audio slot to call on.
//
// Below the CEILING, not below 4000 Hz. The ceiling is the highest offset he
// has actually decoded, and it is the only evidence available about how wide
// his receiver is set — plenty of stations run 2500 Hz. Suggesting 3400 Hz to
// somebody whose passband stops at 2700 is advice to transmit into a filter.
//
// Two passes, and the second is the one that matters on a busy DX. Looking for
// an empty run alone answered "nowhere" exactly when the answer was most
// wanted: a hundred decodes across a 2800 Hz passband leave no run of clear
// bins at all, and the panel drew a full histogram with no advice under it.
// Failing a real gap, the quietest slot is still better than the one the
// operator would have picked by eye.
func suggestOffset(bins []Bin, ceiling int) int {
	if ceiling < 1000 {
		return 0
	}
	count := make(map[int]int, len(bins))
	busy := make(map[int]bool, len(bins)*3)
	for _, b := range bins {
		count[b.OffsetHz] = b.Count
		if b.Count > 0 {
			// The neighbours too: FT8 is 50 Hz wide and the bins are 60, so a
			// signal on a bin edge covers the next one as surely as its own.
			busy[b.OffsetHz], busy[b.OffsetHz-binHz], busy[b.OffsetHz+binHz] = true, true, true
		}
	}
	// From 1000 Hz up: below that is where every default transmit offset sits,
	// so it is the most crowded part of the passband and the least useful advice.
	const low = 1020
	high := (ceiling / binHz) * binHz

	// Pass 1 — the widest clear run, and call from its middle.
	bestStart, bestLen, start, run := -1, 0, -1, 0
	for edge := low; edge <= high; edge += binHz {
		if busy[edge] {
			start, run = -1, 0
			continue
		}
		if start < 0 {
			start = edge
		}
		run++
		if run > bestLen {
			bestStart, bestLen = start, run
		}
	}
	if bestStart >= 0 && bestLen >= 2 {
		return bestStart + bestLen*binHz/2
	}

	// Pass 2 — no clear run: the least busy slot. Walked from the top down, so
	// a tie goes to the higher offset, which is the less crowded half of any
	// passband and the half a pile-up leaves alone.
	quietest, fewest := -1, 1<<30
	for edge := high; edge >= low; edge -= binHz {
		if c := count[edge]; c < fewest {
			quietest, fewest = edge, c
		}
	}
	if quietest < 0 {
		return 0
	}
	// Never past the ceiling: the top bin CONTAINS it, so its middle can sit
	// beyond the highest offset he has been shown to decode — which is the one
	// thing this function exists to avoid.
	if quietest+binHz/2 > ceiling {
		quietest -= binHz
	}
	if quietest < low {
		return 0
	}
	return quietest + binHz/2
}

// bandTag names the band a dial frequency is on, in PSK Reporter's own
// vocabulary ("20m"). Only used by the band-wide scope, to subscribe to one
// band instead of all of them.
func bandTag(hz int64) string {
	khz := hz / 1000
	switch {
	case khz >= 1800 && khz <= 2000:
		return "160m"
	case khz >= 3500 && khz <= 4000:
		return "80m"
	case khz >= 5250 && khz <= 5450:
		return "60m"
	case khz >= 7000 && khz <= 7300:
		return "40m"
	case khz >= 10100 && khz <= 10150:
		return "30m"
	case khz >= 14000 && khz <= 14350:
		return "20m"
	case khz >= 18068 && khz <= 18168:
		return "17m"
	case khz >= 21000 && khz <= 21450:
		return "15m"
	case khz >= 24890 && khz <= 24990:
		return "12m"
	case khz >= 28000 && khz <= 29700:
		return "10m"
	case khz >= 50000 && khz <= 54000:
		return "6m"
	case khz >= 70000 && khz <= 70500:
		return "4m"
	case khz >= 144000 && khz <= 148000:
		return "2m"
	case khz >= 430000 && khz <= 440000:
		return "70cm"
	}
	return ""
}

// ── REST backfill ─────────────────────────────────────────────────────────
//
// The narrow subscription starts empty, and five minutes of waiting is not an
// answer to "should I call this station now". PSK Reporter's query API hands
// back the last quarter hour in one request, so the window is populated before
// the first cycle finishes.
//
// Fetched ONCE per target. The panel polls every second, and a query per poll
// is what gets an application rate-limited off the service for everyone.

type pskrReport struct {
	Sender        string `xml:"senderCallsign,attr"`
	SenderGrid    string `xml:"senderLocator,attr"`
	Receiver      string `xml:"receiverCallsign,attr"`
	ReceiverGrid  string `xml:"receiverLocator,attr"`
	Frequency     string `xml:"frequency,attr"`
	SNR           string `xml:"sNR,attr"`
	Mode          string `xml:"mode,attr"`
	FlowStartSecs string `xml:"flowStartSeconds,attr"`
}

type pskrReports struct {
	XMLName xml.Name     `xml:"receptionReports"`
	Reports []pskrReport `xml:"receptionReport"`
}

// backfill fetches the last quarter hour for a target, in BOTH directions.
//
// Two queries, because the panel asks two questions and the service answers
// them separately: what the target RECEIVED (his pileup, the passband, whether
// he decoded us) and what he TRANSMITTED (who is hearing him, and how much of
// that is near us). The live feed fills both eventually; a target picked ten
// seconds ago has neither, and with the narrow subscription there is nothing in
// the window at all until his own uploader next reports.
//
// Fetched ONCE per target. The panel polls every second, and a query per poll
// is what gets an application rate-limited off the service for everyone.
func (w *Watcher) backfill(target, mode string) {
	w.mu.Lock()
	// Once per target, then no more often than the refresh interval.
	//
	// The live feed alone lags by design: PSK Reporter's uploaders batch their
	// reports, most of them every five minutes, so between two batches the
	// window only holds what happened to have been sent. Asking the history
	// again at that same cadence keeps it as full as a program that has been
	// subscribed for an hour — which is the whole of the difference an operator
	// sees when comparing the two side by side.
	if w.backfilled == target && time.Since(w.backfilledAt) < backfillEvery {
		w.mu.Unlock()
		return
	}
	w.backfilled, w.backfilledAt = target, time.Now()
	w.mu.Unlock()

	got := 0
	for _, dir := range []struct{ param, what string }{
		{"receiverCallsign", "decoded by him"},
		{"senderCallsign", "who is hearing him"},
	} {
		q := url.Values{}
		q.Set(dir.param, target)
		q.Set("mode", mode)
		q.Set("flowStartSeconds", strconv.Itoa(-900))
		q.Set("nolocator", "0")
		// The pskquery5 endpoint rather than retrieve.pskreporter.info: this is
		// the one DXHunter has been using against the live service, and a
		// backfill that silently returns nothing is worse than none at all.
		req, err := http.NewRequest("GET", "https://pskreporter.info/cgi-bin/pskquery5.pl?"+q.Encode(), nil)
		if err != nil {
			continue
		}
		req.Header.Set("User-Agent", "OpsLog (PSK Reporter target analysis)")
		resp, err := (&http.Client{Timeout: 15 * time.Second}).Do(req)
		if err != nil {
			w.cfg.Logf("pskr target: history for %s (%s) unavailable: %v", target, dir.what, err)
			continue
		}
		if resp.StatusCode != http.StatusOK {
			// 503 is the service saying "too often". Worth a line, because the
			// panel then fills at the live feed's pace and looks slow for no
			// visible reason.
			w.cfg.Logf("pskr target: history for %s (%s) refused (HTTP %d)", target, dir.what, resp.StatusCode)
			resp.Body.Close()
			continue
		}
		var rr pskrReports
		err = xml.NewDecoder(resp.Body).Decode(&rr)
		resp.Body.Close()
		if err != nil {
			continue
		}
		got += w.absorb(target, rr.Reports)
	}
	if got > 0 {
		w.cfg.Logf("pskr target: %d recent reports for %s from the history queries", got, target)
	}
}

// absorb adds fetched reports to the window, skipping what the live feed has
// already delivered. Without the check the same report arrives twice — once by
// MQTT, once by query — and every count that is not per-callsign doubles: the
// decode total, and the bars of the passband.
func (w *Watcher) absorb(target string, reports []pskrReport) int {
	now := time.Now()
	w.mu.Lock()
	defer w.mu.Unlock()
	// Still the same target? The operator may have moved on while this was in
	// flight, and dropping a stale answer into the window would attribute one
	// station's pileup to another.
	if w.target != target {
		return 0
	}
	type key struct {
		tx, rx string
		hz     int64
	}
	seen := make(map[key]bool, len(w.spots))
	for i := range w.spots {
		seen[key{w.spots[i].TxCall, w.spots[i].RxCall, w.spots[i].Freq}] = true
	}
	added := 0
	for _, r := range reports {
		hz, _ := strconv.ParseInt(r.Frequency, 10, 64)
		snr, _ := strconv.Atoi(r.SNR)
		k := key{strings.ToUpper(r.Sender), strings.ToUpper(r.Receiver), hz}
		if hz == 0 || seen[k] {
			continue
		}
		at := now
		if secs, err := strconv.ParseInt(r.FlowStartSecs, 10, 64); err == nil {
			switch {
			case secs > 1_000_000_000:
				at = time.Unix(secs, 0) // an absolute time
			case secs < 0:
				at = now.Add(time.Duration(secs) * time.Second) // an age in seconds
			}
		}
		// Stamped with its REAL age, so it ages out of the window on its own and
		// a quarter-hour-old decode is never read as "he heard you just now".
		if at.Before(now.Add(-window)) {
			continue
		}
		seen[k] = true
		w.spots = append(w.spots, spot{
			Freq: hz, Mode: strings.ToUpper(r.Mode), SNR: snr,
			TxCall: k.tx, TxGrid: strings.ToUpper(r.SenderGrid),
			RxCall: k.rx, RxGrid: strings.ToUpper(r.ReceiverGrid),
			at: at,
		})
		added++
	}
	return added
}