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"
)

// 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
		},
	}
}

// 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, CQ decodes only
	DecodeFreqHz int64  // RF frequency (dial + audio offset)
	DecodeSNR    int    // reported SNR (dB)
	DecodeCQ     bool   // the decode was a CQ

	// 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
	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
	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
}

// 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) *Server {
	return &Server{
		cfg:  cfg,
		out:  out,
		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)
	}
}

// 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) {
	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
		}
		// 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[w.ProgramID] = w.FreqHz
			s.mu.Unlock()
		}
		if w.IsDecode {
			s.mu.Lock()
			dial := s.dialHz[w.ProgramID]
			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 = w.Mode
			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)
		// 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 {
			if mhz, err := strconv.ParseFloat(m[1], 64); err == nil && mhz > 0 {
				ev.TuneFreqHz = int64(mhz * 1e6)
			}
			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
		}
		ev.DXCall = strings.ToUpper(parts[len(parts)-1])
	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).
	if ev.DXCall == "" && ev.LoggedADIF == "" && ev.DecodeCall == "" && !ev.ClearCall && ev.TuneFreqHz == 0 {
		return
	}
	select {
	case s.out <- ev:
	default:
		// Drop on backpressure rather than block the read loop.
	}
}

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

	// 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

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

func NewManager(repo *Repo) *Manager {
	return &Manager{
		repo:    repo,
		out:     make(chan Event, 64),
		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)
		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()
	}
	applog.Printf("udp: Reload done — %d server(s) running, %d error(s)", len(m.inbound), len(errs))
	return errs
}

// 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()
	}
}