OpsLog/internal/integrations/udp/server.go

package udp

import (
	"context"
	"encoding/binary"
	"fmt"
	"net"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"golang.org/x/net/ipv4"

	"hamlog/internal/applog"
)

// remoteTuneHz turns a <FREQ> value from a remote-call packet into Hz.
//
// The field has NO agreed unit, and the same sender uses two of them. DXHunter
// documents "<FREQ>10.136" — MHz — but a log from a working station showed it
// echoing "<FREQ>2107400" straight back: the frequency OpsLog had just
// published to it in the N1MM RadioInfo broadcast, whose <Freq> is in units of
// 10 Hz. Read as MHz, that asked the rig for 2 107 400 MHz, and every tune
// request failed with "out of the 11-digit CAT range" from the first second
// after launch.
//
// So the unit is inferred: try each one and keep the first that lands on an
// amateur band. Anything a station is asked to tune to is, by definition, in
// one. Hz is tried before 10 Hz because the one overlap between them — a 40 m
// frequency in Hz reads as a 4 m one in tens of Hz — is far more likely to be
// 40 m. Nothing plausible means nothing is tuned: a wrong band is worse than a
// request that visibly did nothing.
func remoteTuneHz(s string) int64 {
	v, err := strconv.ParseFloat(s, 64)
	if err != nil || v <= 0 {
		return 0
	}
	for _, hz := range []int64{int64(v * 1e6), int64(v * 1e3), int64(v), int64(v * 10)} {
		if bandFromHz(hz) != "" {
			return hz
		}
	}
	// Refusing in silence is how the previous version's failure looked from the
	// outside: a spot clicked in another program, and nothing happening here.
	applog.Printf("udp: remote_call <FREQ>%s is not a frequency in any amateur band in MHz, kHz or Hz — not tuning\n", s)
	return 0
}

// remoteFreqRe / remoteModeRe pull the optional tune request out of a
// ServiceRemoteCall packet: "<FREQ>10.136" (MHz) and "<MODE>FT8". Both accept
// an optional closing tag for proper-XML senders.
var (
	remoteFreqRe = regexp.MustCompile(`(?i)<FREQ>\s*([0-9]+(?:\.[0-9]+)?)`)
	remoteModeRe = regexp.MustCompile(`(?i)<MODE>\s*([A-Z0-9-]+)`)
)

// reusingListenConfig builds a net.ListenConfig that sets SO_REUSEADDR
// (and SO_REUSEPORT on Unix) on the underlying socket before bind. This
// is the only way for two processes to share a UDP port on Windows — Go
// doesn't expose the option directly, but ListenConfig.Control hooks the
// raw socket and lets us call setsockopt.
func reusingListenConfig() net.ListenConfig {
	return net.ListenConfig{
		Control: func(network, address string, c syscall.RawConn) error {
			var opErr error
			err := c.Control(func(fd uintptr) {
				opErr = setSocketReuse(fd)
			})
			if err != nil {
				return err
			}
			return opErr
		},
	}
}

// decodeTime turns WSJT-X's milliseconds-since-midnight into a UTC instant.
//
// The sender gives a time of DAY with no date, so the date comes from our own
// clock — and the two can straddle midnight: a decode stamped 23:59:58 that
// reaches us at 00:00:01 would otherwise be dated a day late and sort to the top
// of the list for the rest of the session. More than half a day apart is read as
// the wrong side of midnight and moved.
func decodeTime(msSinceMidnight uint32) time.Time {
	return decodeTimeAt(msSinceMidnight, time.Now().UTC())
}

// decodeTimeAt is decodeTime with the clock passed in.
//
// Split out so the midnight rule can be tested at midnight instead of at
// whatever time the suite happens to run: the test used to read the real clock
// and failed every afternoon, because a 23:59:58 stamp seen at 17:00 is hours in
// the future and no rule can make it otherwise. A test that fails for half the
// day teaches everyone to ignore the suite.
func decodeTimeAt(msSinceMidnight uint32, now time.Time) time.Time {
	now = now.UTC()
	midnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
	at := midnight.Add(time.Duration(msSinceMidnight) * time.Millisecond)
	switch {
	case at.Sub(now) > 12*time.Hour:
		at = at.AddDate(0, 0, -1) // stamped late yesterday, arrived after midnight
	case now.Sub(at) > 12*time.Hour:
		at = at.AddDate(0, 0, 1) // stamped just after midnight, our clock still on the old day
	}
	return at
}

// Event is what a Server emits to its consumer for every parsed packet.
// At most one of the fields is populated per event.
type Event struct {
	ConfigID int64
	Service  ServiceType
	Source   string // remote addr that sent the packet, for diagnostics

	DXCall     string // ServiceWSJT (Status) or ServiceRemoteCall
	DXGrid     string // ServiceWSJT (Status)
	Mode       string // ServiceWSJT (Status/Decode)
	FreqHz     int64  // ServiceWSJT (Status)
	LoggedADIF string // ServiceWSJT (LoggedADIF), ServiceADIF or ServiceN1MM

	// A WSJT-X Decode (heard station) to render on the panadapter.
	DecodeCall   string // transmitting (DE) callsign
	DecodeGrid   string // 4-char grid: from a CQ, or a reply answering with its locator
	DecodeFreqHz int64  // RF frequency (dial + audio offset)
	DecodeSNR    int    // reported SNR (dB)
	DecodeCQ     bool   // the decode was a CQ
	DecodeMsg    string // the decoded line as printed ("CQ K1ABC FN42")
	// DecodeAt is the decode's own UTC timestamp, rebuilt from the sender's
	// milliseconds-since-midnight. It is what groups decodes into T/R periods:
	// a period's worth arrives in one burst, so arrival time would put them all
	// in whichever slot the burst happened to land in.
	DecodeAt time.Time
	// DecodeTRPeriod is the transmit/receive period in seconds, from the last
	// Status of the same program (15 = FT8). 0 when the sender never said.
	DecodeTRPeriod int
	DecodeDial     int64 // dial frequency the decode was heard on, for the band
	DecodeOffAir   bool  // decoded from a file rather than off the air
	// The three fields below are shown in the panel AND replayed verbatim when
	// answering the station — WSJT-X matches a Reply against its own decode list,
	// so every one has to go back exactly as it came.
	DecodeDT      float64 // seconds into the slot the transmission started
	DecodeAudioHz int64   // audio offset inside the passband
	DecodeMs      uint32  // the decode's raw ms-since-midnight, as sent
	DecodeLowConf bool
	// DecodeModeRaw is the mode field EXACTLY as the Decode carried it — the
	// one-character marker ("~", "+"), not the resolved name in Mode.
	//
	// Both are needed and they are not interchangeable. Mode is what the log and
	// the status resolver want. This is what a Reply must echo: the receiving
	// application matches a Reply against its own decode list field for field,
	// and JTDX rejects one whose mode reads "FT8" where it decoded "~" — the
	// click is accepted, nothing is transmitted, and nothing is reported.
	DecodeModeRaw string
	// DecodeMsgRaw is the message as sent, untrimmed — see DecodeModeRaw.
	DecodeMsgRaw string
	// DecodeIsNew is false on the history a Replay resends: display-only lines.
	DecodeIsNew bool
	// ProgramID is the sending application's own id ("WSJT-X", "MSHV", or
	// "WSJT-X - 2" for a second instance started with --rig-name). It is what
	// tells two receivers apart on one multicast group — and it is the address a
	// Reply message would have to be sent back to, so it is carried even though
	// nothing replies yet.
	ProgramID string

	// TxMessage is what the operator's digital app is sending, with Transmitting
	// true while the carrier is actually up. From Status, so ~1 Hz.
	TxMessage    string
	Transmitting bool
	// TxEnabled is the sender's Enable Tx toggle — see WSJTEvent.TxEnabled. The
	// watchdog clears it, which is the only reliable sign an exchange was
	// abandoned rather than merely paused between overs.
	TxEnabled bool
	DECall    string // the operator's own call, as the digital app knows it

	// ClearCall is set when a WSJT/JTDX/MSHV Status message reports an EMPTY DX
	// Call after previously reporting one — i.e. the operator cleared the call in
	// the digital app. OpsLog clears its entry to match.
	ClearCall bool

	// TuneFreqHz / TuneMode carry an explicit "tune the radio" request embedded
	// in a ServiceRemoteCall packet (DXHunter spot click sends
	// "<CALLSIGN>VP5G<FREQ>10.136<MODE>FT8"). Zero/empty = no tune requested —
	// the packet only fills the entry callsign, exactly as before.
	TuneFreqHz int64
	TuneMode   string
}

// Server is a single inbound UDP listener.
type Server struct {
	cfg Config
	// mgr is the owning Manager, so a listener can reach the outbound rows.
	// The relay is the only thing that needs it.
	mgr     *Manager
	conn    *net.UDPConn
	out     chan<- Event
	stop    chan struct{}
	done    chan struct{}
	stopped bool
	mu      sync.Mutex

	// WSJT: dial frequency from the last Status, added to Decode audio offsets —
	// keyed by the WSJT-X instance id, NOT one value per listener. Two instances
	// share one multicast group, so a single value was overwritten by whichever
	// sent Status last: decodes from the 50.313 receiver were placed on the
	// 50.400 panadapter. WSJT-X requires --rig-name for a second instance, so the
	// id is distinct whenever there is more than one.
	dialHz map[string]int64
	// trPeriod is the T/R period (seconds) from each program's last Status —
	// what tells a decode which slot it belongs to.
	trPeriod map[string]int
	// lastFrom is the address each program's packets arrive from — where a Reply
	// has to be sent. See SendReply.
	lastFrom map[string]*net.UDPAddr
	// onNewInstance fires (off the read loop) the first time a program id is
	// heard on this listener — the hook the startup replay hangs from.
	onNewInstance func(programID string)
	// instLabel names each running application, keyed by id AND sending address.
	//
	// WSJT-X requires --rig-name for a second instance, so its ids differ. MSHV
	// does not: two copies both call themselves "MSHV", and everything below was
	// keyed on that name alone — so the second copy's Status overwrote the
	// first's dial frequency, and decodes from the 20 m instance were placed at
	// the LF instance's frequency. Reported as FT8 decodes on 14.095 labelled
	// 2190 m, which is exactly 136 kHz plus an audio offset.
	//
	// The label is what the rest of OpsLog sees: the plain id for the first
	// instance, then "MSHV #2", "MSHV #3"… so the decodes panel can still tell
	// them apart on screen.
	instLabel map[string]string
	// instSeq counts how many distinct instances have claimed each id.
	instSeq map[string]int
	// lastMode is the mode NAME from each program's last Status, used to resolve
	// a Decode's one-character mode marker.
	lastMode map[string]string
	lastDX   string // WSJT: last non-empty DX Call seen, to detect a clear

	// badPkts counts datagrams this listener could not parse, so the diagnostic
	// dump below stays bounded. A misconfigured port is not a one-off: the
	// sender that produced "bad magic 0x3132372e" put out ~150 packets a second,
	// which fills the whole rotating log with the same line and buries the
	// evidence of anything else.
	badPkts int

	// drops counts events discarded because the consumer could not keep up —
	// see the select at the foot of handle(). They used to vanish in silence,
	// which made a missing decode indistinguishable from one the sender never
	// broadcast: an operator comparing the panel against JTDX side by side was
	// the only detector we had. A period's decodes arrive as one burst, so this
	// is exactly when it happens.
	drops     int
	lastDropL time.Time
}

// maxBadPktDumps is how many unparseable datagrams a listener describes in full
// before going quiet. Enough to identify the sender and the payload; few enough
// that a permanently misconfigured port costs a handful of lines, not a log.
const maxBadPktDumps = 5

// describePacket renders a datagram for the log: its size, a printable preview
// and the first bytes in hex.
//
// Both forms, deliberately. "bad magic 0x3132372e" is already readable as ASCII
// "127." to someone who thinks to decode it — and that one fact (the sender is
// emitting text, not WSJT-X binary) is the whole diagnosis. The hex stays for
// the case where the payload really is binary and the preview shows nothing.
func describePacket(pkt []byte) string {
	const maxShown = 96
	head := pkt
	if len(head) > maxShown {
		head = head[:maxShown]
	}
	var text, hex strings.Builder
	for _, b := range head {
		if b >= 0x20 && b < 0x7f {
			text.WriteByte(b)
		} else {
			text.WriteByte('.')
		}
		fmt.Fprintf(&hex, "%02x ", b)
	}
	more := ""
	if len(pkt) > maxShown {
		more = "…"
	}
	return fmt.Sprintf("%d bytes | text %q%s | hex %s%s", len(pkt), text.String(), more, strings.TrimSpace(hex.String()), more)
}

func newServer(cfg Config, out chan<- Event, mgr *Manager) *Server {
	return &Server{
		cfg:           cfg,
		out:           out,
		mgr:           mgr,
		onNewInstance: mgr.onNewInstance,
		stop:          make(chan struct{}),
		done:          make(chan struct{}),
	}
}

func (s *Server) start() error {
	var conn *net.UDPConn
	if s.cfg.Multicast {
		group := strings.TrimSpace(s.cfg.MulticastGroup)
		if group == "" {
			return fmt.Errorf("multicast enabled but group address is empty")
		}
		groupIP := net.ParseIP(group)
		if groupIP == nil {
			return fmt.Errorf("bad multicast group %q", group)
		}
		gaddr := &net.UDPAddr{IP: groupIP, Port: s.cfg.Port}
		// Bind to INADDR_ANY:port so the kernel will forward packets
		// addressed to the multicast group from any interface. Then
		// JoinGroup() on every up & multicast-capable interface — Windows
		// won't route multicast through interfaces we haven't explicitly
		// joined, and the "default" interface picked by
		// net.ListenMulticastUDP isn't always the one MSHV/WSJT sends on.
		// ListenConfig with SO_REUSEADDR lets us share the port with
		// Log4OM / other listeners already bound to 2237.
		lc := reusingListenConfig()
		pc, err := lc.ListenPacket(context.Background(), "udp4", fmt.Sprintf("0.0.0.0:%d", s.cfg.Port))
		if err != nil {
			return fmt.Errorf("listen :%d for multicast: %w", s.cfg.Port, err)
		}
		c, ok := pc.(*net.UDPConn)
		if !ok {
			_ = pc.Close()
			return fmt.Errorf("internal: ListenPacket returned %T not *net.UDPConn", pc)
		}
		p := ipv4.NewPacketConn(c)
		ifaces, _ := net.Interfaces()
		joined := 0
		for _, ifi := range ifaces {
			if ifi.Flags&net.FlagUp == 0 || ifi.Flags&net.FlagMulticast == 0 {
				continue
			}
			if err := p.JoinGroup(&ifi, gaddr); err != nil {
				applog.Printf("udp: [%s] join %s on %s: %v\n", s.cfg.Name, gaddr.IP, ifi.Name, err)
				continue
			}
			joined++
		}
		if joined == 0 {
			_ = c.Close()
			return fmt.Errorf("couldn't join multicast %s on any interface", gaddr.IP)
		}
		conn = c
		applog.Printf("udp: [%s] listening on multicast %s on %d interface(s) (service=%s)\n",
			s.cfg.Name, gaddr, joined, s.cfg.ServiceType)
	} else {
		lc := reusingListenConfig()
		pc, err := lc.ListenPacket(context.Background(), "udp4", fmt.Sprintf("0.0.0.0:%d", s.cfg.Port))
		if err != nil {
			return fmt.Errorf("listen udp :%d: %w", s.cfg.Port, err)
		}
		c, ok := pc.(*net.UDPConn)
		if !ok {
			_ = pc.Close()
			return fmt.Errorf("internal: ListenPacket returned %T not *net.UDPConn", pc)
		}
		conn = c
		applog.Printf("udp: [%s] listening on unicast :%d (service=%s)\n", s.cfg.Name, s.cfg.Port, s.cfg.ServiceType)
	}
	s.conn = conn
	go s.run()
	return nil
}

func (s *Server) run() {
	defer close(s.done)
	buf := make([]byte, 64*1024)
	for {
		select {
		case <-s.stop:
			return
		default:
		}
		_ = s.conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
		n, remote, err := s.conn.ReadFromUDP(buf)
		if err != nil {
			if ne, ok := err.(net.Error); ok && ne.Timeout() {
				continue
			}
			// Closed by stop(): exit silently.
			return
		}
		if n == 0 {
			continue
		}
		pkt := make([]byte, n)
		copy(pkt, buf[:n])
		go s.handle(pkt, remote)
	}
}

// instanceLabel returns the name for one running application, allocating it on
// first sight. Caller holds s.mu.
//
// The address is part of the identity because the id is not enough: two copies
// of MSHV send the same id from different sockets, and the two are different
// radios on different bands. The port is included — a program keeps its socket
// for as long as it runs, which is exactly the lifetime this has to be stable
// over.
func (s *Server) instanceLabel(id string, remote *net.UDPAddr) string {
	if id == "" {
		return ""
	}
	if remote == nil {
		return id
	}
	key := id + "|" + remote.String()
	if s.instLabel == nil {
		s.instLabel = map[string]string{}
		s.instSeq = map[string]int{}
	}
	if lbl, ok := s.instLabel[key]; ok {
		return lbl
	}
	s.instSeq[id]++
	lbl := id
	if n := s.instSeq[id]; n > 1 {
		lbl = fmt.Sprintf("%s #%d", id, n)
		applog.Printf("udp: [%s] a second application calls itself %q (from %s) — it will be shown as %q",
			s.cfg.Name, id, remote, lbl)
	}
	s.instLabel[key] = lbl
	return lbl
}

// logBadPacket reports a datagram this listener could not parse, with enough of
// it to identify the sender — then falls silent.
//
// The point is the SENDER: an unparseable packet on a WSJT port almost always
// means another program is broadcasting on it, or the service type is wrong for
// what is actually arriving. The remote address names the culprit, and the
// payload preview says what it really is. Neither was logged before, so the
// operator saw only a magic number repeated a few hundred times a second.
func (s *Server) logBadPacket(kind string, remote *net.UDPAddr, pkt []byte, err error) {
	s.mu.Lock()
	s.badPkts++
	n := s.badPkts
	s.mu.Unlock()
	if n > maxBadPktDumps {
		return
	}
	applog.Printf("udp: [%s] %s parse error from %s: %v — %s\n",
		s.cfg.Name, kind, remote, err, describePacket(pkt))
	if n == maxBadPktDumps {
		applog.Printf("udp: [%s] further unparseable packets on port %d will not be logged — "+
			"check that the sender belongs on this port and that the service type matches\n",
			s.cfg.Name, s.cfg.Port)
	}
}

// noteIgnoredADIF reports a payload the ADIF listener could not use, ONCE per
// cause and then quietly.
//
// This used to log a line per datagram. An operator's log arrived with four
// hundred identical "ADIF payload ignored" lines and nothing else legible: two
// inbound listeners were configured on the SAME multicast group and port
// (239.255.0.1:2237, one WSJT and one ADIF), so every WSJT-X decode packet was
// delivered to both, and the ADIF one rejected each of them in writing. An FT8
// cycle is dozens of decodes every fifteen seconds.
//
// So the WSJT-X magic number is checked here. When it matches, the payload is
// not "chatter" — it is a specific, fixable misconfiguration, and saying which
// is the difference between a log the operator can act on and one they cannot.
func (s *Server) noteIgnoredADIF(pkt []byte) {
	s.mu.Lock()
	s.badPkts++
	n := s.badPkts
	s.mu.Unlock()
	if n > maxBadPktDumps {
		return
	}
	if len(pkt) >= 4 && binary.BigEndian.Uint32(pkt[:4]) == wsjtMagic {
		applog.Printf("udp: [%s] this is WSJT-X traffic, not ADIF — port %d is shared with a WSJT source; "+
			"give the ADIF forwarder (JTAlert / GridTracker) its own port, or set this listener's service to WSJT\n",
			s.cfg.Name, s.cfg.Port)
		// One line is the whole diagnosis: nothing is gained by counting to five.
		s.mu.Lock()
		s.badPkts = maxBadPktDumps + 1
		s.mu.Unlock()
		return
	}
	applog.Printf("udp: [%s] ADIF payload ignored (no <call:>/<eor>) — %s\n", s.cfg.Name, describePacket(pkt))
	if n == maxBadPktDumps {
		applog.Printf("udp: [%s] further unusable payloads on port %d will not be logged\n",
			s.cfg.Name, s.cfg.Port)
	}
}

func (s *Server) handle(pkt []byte, remote *net.UDPAddr) {
	// Relay FIRST, and unconditionally: before parsing, and whatever the parse
	// then makes of it. A datagram this build cannot decode is still one the
	// application downstream may understand, and a relay that forwards only what
	// it understood drops exactly the fields it has yet to learn about.
	//
	// WSJT listeners only. The other inbound services are text protocols with no
	// second consumer to speak of, and relaying an ADIF record twice would log
	// the contact twice.
	if s.cfg.ServiceType == ServiceWSJT && s.mgr != nil {
		s.mgr.RelayInbound(pkt, s.cfg.Port)
	}
	ev := Event{ConfigID: s.cfg.ID, Service: s.cfg.ServiceType, Source: remote.String()}
	switch s.cfg.ServiceType {
	case ServiceWSJT:
		w, ok, err := ParseWSJT(pkt)
		if err != nil {
			s.logBadPacket("WSJT", remote, pkt, err)
			return
		}
		if !ok {
			return
		}
		// Where this application's packets come from, so a Reply can be sent back
		// to it. Per PROGRAM, not per listener: two receivers share one multicast
		// group, and a reply must reach the one that heard the station — and it
		// must go to the sender's own address, never to the group.
		s.mu.Lock()
		inst := s.instanceLabel(w.ProgramID, remote)
		newInstance := false
		if inst != "" && remote != nil {
			if s.lastFrom == nil {
				s.lastFrom = map[string]*net.UDPAddr{}
			}
			if _, known := s.lastFrom[inst]; !known {
				newInstance = true
			}
			s.lastFrom[inst] = remote
		}
		onNew := s.onNewInstance
		s.mu.Unlock()
		// A program just heard for the first time this session: tell the app, so
		// it can ask for a replay of the decodes already on that program's screen.
		if newInstance && onNew != nil {
			go onNew(inst)
		}
		// Status carries the current dial frequency; remember it so Decode audio
		// offsets can be turned into RF frequencies for the panadapter.
		if w.FreqHz > 0 && !w.IsDecode {
			s.mu.Lock()
			if s.dialHz == nil {
				s.dialHz = map[string]int64{}
			}
			s.dialHz[inst] = w.FreqHz
			// The T/R period travels with Status, and a decode has to be told
			// which slot it belongs to — so it is remembered per program the
			// same way the dial is.
			if w.TRPeriod > 0 {
				if s.trPeriod == nil {
					s.trPeriod = map[string]int{}
				}
				s.trPeriod[inst] = w.TRPeriod
			}
			// The mode NAME, which only Status carries: a Decode gives the
			// one-character marker instead. See DecodeModeName.
			if w.Mode != "" {
				if s.lastMode == nil {
					s.lastMode = map[string]string{}
				}
				s.lastMode[inst] = w.Mode
			}
			s.mu.Unlock()
		}
		if !w.IsDecode && (w.TxMessage != "" || w.DECall != "") {
			// What the operator is sending, and to whom. Carried on every Status,
			// so the consumer sees it change as the QSO progresses. The program id
			// travels with it because a second receiver has a transmit state of
			// its own.
			ev.TxMessage = w.TxMessage
			ev.Transmitting = w.Transmitting
			ev.DECall = w.DECall
			ev.ProgramID = inst
		}
		if w.IsDecode {
			s.mu.Lock()
			dial := s.dialHz[inst]
			tr := s.trPeriod[inst]
			statusMode := s.lastMode[inst]
			s.mu.Unlock()
			if dial <= 0 {
				// No Status from THIS instance yet. Guessing with another
				// instance's dial is what produced the wrong-panadapter spots,
				// so drop the decode and wait — Status arrives every second.
				return
			}
			ev.DecodeCall = w.DecodeCall
			ev.DecodeGrid = w.DecodeGrid
			ev.DecodeFreqHz = dial + w.DeltaFreqHz
			ev.DecodeSNR = w.SNR
			ev.DecodeCQ = w.IsCQ
			ev.Mode = DecodeModeName(w.Mode, statusMode)
			ev.DecodeModeRaw = w.Mode
			ev.DecodeMsg = w.DecodeMsg
			ev.DecodeMsgRaw = w.DecodeMsgRaw
			ev.DecodeIsNew = w.DecodeIsNew
			ev.DecodeAt = decodeTime(w.DecodeMsSinceMidnight)
			ev.DecodeTRPeriod = tr
			ev.DecodeDial = dial
			ev.DecodeOffAir = w.OffAir
			ev.ProgramID = inst
			ev.DecodeDT = w.DeltaTime
			ev.DecodeAudioHz = w.DeltaFreqHz
			ev.DecodeMs = w.DecodeMsSinceMidnight
			ev.DecodeLowConf = w.LowConfidence
			break
		}
		// Only a logged QSO is worth a line — WSJT-X/MSHV send a Status packet
		// every second, and logging each one buried the rest of the file.
		if len(w.LoggedADIF) > 0 {
			applog.Printf("udp: [%s] WSJT QSO logged: prog=%q dx_call=%q grid=%q mode=%q freq=%d adif_len=%d\n",
				s.cfg.Name, w.ProgramID, w.DXCall, w.DXGrid, w.Mode, w.FreqHz, len(w.LoggedADIF))
		}
		ev.DXCall = w.DXCall
		ev.DXGrid = w.DXGrid
		ev.Mode = w.Mode
		ev.FreqHz = w.FreqHz
		ev.LoggedADIF = w.LoggedADIF
		// A Status with an empty DX Call, right after one that had a call, means the
		// operator cleared it in WSJT-X / JTDX / MSHV. Fire ONE clear (tracked per
		// server) — an idle app sends empty Status every second, and we must not
		// re-clear (which would fight a manual entry) on each of those.
		s.mu.Lock()
		prev := s.lastDX
		s.lastDX = w.DXCall
		s.mu.Unlock()
		if w.DXCall == "" && prev != "" {
			ev.ClearCall = true
		}
	case ServiceADIF:
		// JTAlert / GridTracker forward a text ADIF record after a QSO is
		// logged. Guard against keep-alive / non-ADIF chatter on the socket:
		// only forward payloads that actually carry a callsign field and a
		// record terminator.
		text := string(pkt)
		low := strings.ToLower(text)
		if !strings.Contains(low, "<call:") || !strings.Contains(low, "<eor") {
			s.noteIgnoredADIF(pkt)
			return
		}
		ev.LoggedADIF = text
	case ServiceRemoteCall:
		// Common payload shapes seen in the wild:
		//   "F4XYZ"                       (bare callsign)
		//   "CALL F4XYZ"                  (text prefix)
		//   "<CALLSIGN>F4XYZ<CALLSIGN>"   (DXHunter-style tags)
		//   "<CALLSIGN>F4XYZ</CALLSIGN>"  (proper XML)
		//   "<CALLSIGN>VP5G<FREQ>10.136<MODE>FT8"  (DXHunter with CAT tune)
		// Strip every angle-bracket tag, normalise whitespace, take the
		// last non-empty token. Upper-case for downstream consistency.
		text := string(pkt)
		// NEVER act on an N1MM RadioInfo datagram.
		//
		// It is not a spot click, it is a radio TELLING the world where it is —
		// and on a station where an inbound remote-call row and an outbound
		// RadioInfo row share a port, the one OpsLog just sent arrives straight
		// back on the loopback. The tag-stripping below then reads its last token,
		// <ActiveRadioNr>1</ActiveRadioNr>, as the callsign "1", and <Freq> as a
		// tune request — for the frequency the rig is already on.
		//
		// That was harmless only for as long as <Freq> was misread: in tens of Hz
		// it looks like a wild number, every tune failed "out of the CAT range",
		// and the loop died there. Reading the unit correctly closed it. With
		// WSJT-X/JTDX "Fake It", which shifts the dial for each over, every
		// transmission then produced a burst of sets echoing between OpsLog and
		// itself until the rig stopped answering IF; and the shared CAT link
		// dropped. Reported as Fake It causing CAT disconnections.
		if low := strings.ToLower(text); strings.Contains(low, "<radioinfo") {
			return
		}
		// Optional tune request: <FREQ>MHz and <MODE>str ride along with the
		// callsign so a DXHunter spot click can drive OpsLog's CAT. Extract
		// (and cut) them BEFORE the generic tag-stripping below, which would
		// otherwise leave their values as stray tokens and corrupt the
		// "last token = callsign" heuristic.
		if m := remoteFreqRe.FindStringSubmatch(text); m != nil {
			ev.TuneFreqHz = remoteTuneHz(m[1])
			text = strings.Replace(text, m[0], " ", 1)
		}
		if m := remoteModeRe.FindStringSubmatch(text); m != nil {
			ev.TuneMode = strings.ToUpper(m[1])
			text = strings.Replace(text, m[0], " ", 1)
		}
		// Drop every <...> tag (open or close) — works for both
		// <CALLSIGN>...<CALLSIGN> and <CALLSIGN>...</CALLSIGN>.
		for {
			start := strings.IndexByte(text, '<')
			if start < 0 {
				break
			}
			end := strings.IndexByte(text[start:], '>')
			if end < 0 {
				break
			}
			text = text[:start] + " " + text[start+end+1:]
		}
		text = strings.TrimSpace(text)
		parts := strings.Fields(text)
		if len(parts) == 0 {
			return
		}
		call := strings.ToUpper(parts[len(parts)-1])
		// A callsign has a letter in it. Without this, any status XML that ends
		// in a number is read as a station — the RadioInfo above was exactly
		// that, and refusing it by shape as well as by name means the next
		// program to broadcast its state on this port cannot drive the rig
		// either.
		if !strings.ContainsAny(call, "ABCDEFGHIJKLMNOPQRSTUVWXYZ") {
			return
		}
		ev.DXCall = call
	case ServiceN1MM:
		adifText, ok, err := ParseN1MM(pkt)
		if err != nil {
			s.logBadPacket("N1MM", remote, pkt, err)
			return
		}
		if !ok {
			applog.Printf("udp: [%s] N1MM datagram ignored (not a loggable contact)\n", s.cfg.Name)
			return
		}
		applog.Printf("udp: [%s] N1MM contact decoded (%d bytes ADIF)\n", s.cfg.Name, len(adifText))
		ev.LoggedADIF = adifText
	default:
		return
	}
	// Empty events are useless; skip — EXCEPT a clear signal, which is meant to be
	// empty (the DX Call was cleared in the digital app), and a tune-only
	// request (freq with no callsign).
	// TxMessage rides on Status, which also carries the DX call — but a Status
	// with an empty DX call and a live transmit message (calling CQ) used to be
	// dropped here, and that is exactly the message the decodes panel needs.
	if ev.DXCall == "" && ev.LoggedADIF == "" && ev.DecodeCall == "" && ev.TxMessage == "" && !ev.ClearCall && ev.TuneFreqHz == 0 {
		return
	}
	select {
	case s.out <- ev:
	default:
		// Drop on backpressure rather than block the read loop — but say so.
		s.noteDrop(ev)
	}
}

// noteDrop reports events lost to a full queue, first one immediately and then
// at most one line a minute with the running total.
//
// Rate-limited because the condition is self-sustaining: a consumer that fell
// behind on one period is behind for the next, and a line per lost decode would
// bury the rest of the log under the symptom. The event's own description goes
// in, because "a decode was lost" and "a Status was lost" are different faults.
func (s *Server) noteDrop(ev Event) {
	what := "event"
	switch {
	case ev.DecodeCall != "":
		what = "decode from " + ev.DecodeCall
	case ev.LoggedADIF != "":
		what = "logged QSO"
	case ev.TxMessage != "" || ev.DECall != "":
		what = "transmit status"
	}
	s.mu.Lock()
	s.drops++
	n := s.drops
	quiet := n > 1 && time.Since(s.lastDropL) < time.Minute
	if !quiet {
		s.lastDropL = time.Now()
	}
	s.mu.Unlock()
	if quiet {
		return
	}
	applog.Printf("udp: [%s] queue full — dropped %s (%d lost this session); "+
		"the decodes panel is missing what the sender broadcast", s.cfg.Name, what, n)
}

func (s *Server) close() {
	s.mu.Lock()
	if s.stopped {
		s.mu.Unlock()
		return
	}
	s.stopped = true
	stop, done, conn := s.stop, s.done, s.conn
	s.mu.Unlock()
	if conn != nil {
		_ = conn.Close()
	}
	if stop != nil {
		close(stop)
	}
	if done != nil {
		<-done
	}
}

// ── Outbound emitter ──────────────────────────────────────────────────

// SendUDP sends payload to dst (host:port). Unicast or directed broadcast.
// Returns the error from the write; the connection is closed before return.
func SendUDP(dst string, payload []byte) error {
	conn, err := net.Dial("udp4", dst)
	if err != nil {
		return fmt.Errorf("dial %s: %w", dst, err)
	}
	defer conn.Close()
	_ = conn.SetWriteDeadline(time.Now().Add(2 * time.Second))
	_, err = conn.Write(payload)
	return err
}

// ── Manager ───────────────────────────────────────────────────────────

// Manager owns every inbound Server and exposes a helper to emit on
// outbound connections at QSO-save time. It reloads from the Repo on
// demand (after a CRUD change in the Settings panel).
type Manager struct {
	repo *Repo
	out  chan Event

	// onNewInstance is copied onto every inbound listener as it starts; see
	// Server.onNewInstance.
	onNewInstance func(programID string)

	// noADIFOnce keeps the "nothing to forward to" note to one line a session
	// rather than one per QSO logged.
	noADIFOnce sync.Once

	// httpFailAt throttles the complaint from a custom URL row, per row: a
	// switch that is off answers the same way on every band change.
	httpFailMu sync.Mutex
	httpFailAt map[int64]time.Time

	// relayWarned keeps a relay row's complaint (a loop target, a dead
	// destination) to one line a session. Keyed by row id for the loop warning
	// and by -id-1 for the send failure, so one row can say each once.
	// Without this a busy band writes hundreds of identical lines a minute.
	relayWarnMu sync.Mutex
	relayWarned map[int64]bool

	mu       sync.Mutex
	inbound  map[int64]*Server
	outbound []Config
}

func NewManager(repo *Repo) *Manager {
	return &Manager{
		repo: repo,
		// 256, not 64: a period delivers twenty-odd decodes in one burst while
		// Status keeps arriving, and the consumer does real work per event. The
		// depth is headroom for that burst — the drop counter above is what says
		// whether it was enough.
		out:     make(chan Event, 256),
		inbound: map[int64]*Server{},
	}
}

// Events returns the channel inbound parsed events are delivered on.
// The app exposes these as Wails events.
func (m *Manager) Events() <-chan Event { return m.out }

// Reload restarts every server based on the current Repo contents.
// Existing servers are stopped, the snapshot is rebuilt from scratch.
// Errors on individual rows are logged via the returned slice; the
// caller can surface them in the UI.
func (m *Manager) Reload(ctx context.Context) []string {
	applog.Printf("udp: Reload() called")
	m.mu.Lock()
	old := m.inbound
	m.inbound = map[int64]*Server{}
	m.outbound = nil
	m.mu.Unlock()

	for _, s := range old {
		s.close()
	}

	cfgs, err := m.repo.List(ctx)
	if err != nil {
		applog.Printf("udp: Reload list failed: %v", err)
		return []string{fmt.Sprintf("load udp configs: %v", err)}
	}
	applog.Printf("udp: Reload found %d config(s) in DB", len(cfgs))
	var errs []string
	for _, c := range cfgs {
		applog.Printf("udp: cfg id=%d name=%q dir=%s service=%s port=%d mcast=%v group=%q enabled=%v",
			c.ID, c.Name, c.Direction, c.ServiceType, c.Port, c.Multicast, c.MulticastGroup, c.Enabled)
		if !c.Enabled {
			continue
		}
		if c.Direction == Outbound {
			m.mu.Lock()
			m.outbound = append(m.outbound, c)
			m.mu.Unlock()
			continue
		}
		srv := newServer(c, m.out, m)
		if err := srv.start(); err != nil {
			applog.Printf("udp: start %q failed: %v", c.Name, err)
			errs = append(errs, fmt.Sprintf("%s: %v", c.Name, err))
			continue
		}
		m.mu.Lock()
		m.inbound[c.ID] = srv
		m.mu.Unlock()
	}
	warnSharedPorts(cfgs)
	applog.Printf("udp: Reload done — %d server(s) running, %d error(s)", len(m.inbound), len(errs))
	return errs
}

// warnSharedPorts names an inbound and an outbound row sitting on the same
// port, because that is a loop: what OpsLog sends there, OpsLog receives.
//
// It is how a station ended up re-tuning its own rig from its own RadioInfo
// broadcasts. The parser refuses that particular payload now, but the
// arrangement stays wrong for anything else that lands on the port, and it is
// invisible in a settings panel that shows one row at a time.
func warnSharedPorts(cfgs []Config) {
	in := map[int]string{}
	for _, c := range cfgs {
		if c.Enabled && c.Direction != Outbound {
			in[c.Port] = c.Name
		}
	}
	for _, c := range cfgs {
		if !c.Enabled || c.Direction != Outbound {
			continue
		}
		if name, ok := in[c.Port]; ok {
			applog.Printf("udp: %q sends on port %d and %q listens on it — OpsLog will receive its own messages there; give one of the two another port",
				c.Name, c.Port, name)
		}
	}
}

// Outbound returns the active outbound configs matching a service type.
// Used by the QSO save path to push notifications to listeners.
func (m *Manager) Outbound(service ServiceType) []Config {
	m.mu.Lock()
	defer m.mu.Unlock()
	var out []Config
	for _, c := range m.outbound {
		if c.ServiceType == service {
			out = append(out, c)
		}
	}
	return out
}

// StopAll closes every running server. Called at app shutdown.
func (m *Manager) StopAll() {
	m.mu.Lock()
	old := m.inbound
	m.inbound = map[int64]*Server{}
	m.outbound = nil
	m.mu.Unlock()
	for _, s := range old {
		s.close()
	}
}

// SetOnNewInstance installs the first-sighting hook. Call before Reload so
// listeners are born with it.
func (m *Manager) SetOnNewInstance(fn func(programID string)) {
	m.mu.Lock()
	m.onNewInstance = fn
	m.mu.Unlock()
}