// Package cluster provides a multi-server DX cluster client (telnet) — // connects concurrently to several AR-Cluster / CC-Cluster / DXSpider // nodes, logs in with the operator's callsign, optionally sends an init // command list, and streams DX spots back to the UI via a callback. // // Spot parsing is tolerant of the dozens of slight format variations // between cluster flavours (the prompt, the trailing locator, the time // format). Anything that doesn't match the spot regex is treated as // banner/chat noise and ignored, not surfaced as an error. package cluster import ( "bufio" "fmt" "net" "regexp" "strconv" "strings" "sync" "time" "hamlog/internal/applog" ) // ServerConfig is the persisted shape of one cluster node. Mirrors the // columns of the cluster_servers table; the frontend SettingsPanel // pushes one of these per row. type ServerConfig struct { ID int64 `json:"id"` Name string `json:"name"` Host string `json:"host"` Port int `json:"port"` LoginOverride string `json:"login_override"` Password string `json:"password,omitempty"` InitCommands string `json:"init_commands"` // newline-separated, sent post-login Enabled bool `json:"enabled"` SortOrder int `json:"sort_order"` } // Spot is a single DX spot as parsed from the cluster stream. // Country/Continent are filled by the caller (app.go) before the spot // is emitted to the UI, so the table never has empty country cells // flickering in for a few hundred ms. type Spot struct { SourceID int64 `json:"source_id"` // ID of the cluster server this came from SourceName string `json:"source_name"` // display name (handy in the UI when multiple servers) Spotter string `json:"spotter"` // DE field DXCall string `json:"dx_call"` // the DX station heard FreqKHz float64 `json:"freq_khz"` FreqHz int64 `json:"freq_hz"` Band string `json:"band,omitempty"` Comment string `json:"comment,omitempty"` Locator string `json:"locator,omitempty"` // spotter grid (optional) TimeUTC string `json:"time_utc,omitempty"` Country string `json:"country,omitempty"` // DXCC entity name (cty.dat) Continent string `json:"continent,omitempty"` // 2-letter continent CQZone int `json:"cqz,omitempty"` // DXCC entity CQ zone ITUZone int `json:"ituz,omitempty"` // DXCC entity ITU zone DistanceKm int `json:"distance_km,omitempty"` // great-circle km from operator's grid ShortPath int `json:"sp_deg,omitempty"` // azimuth (deg) short path from operator LongPath int `json:"lp_deg,omitempty"` // azimuth (deg) long path = SP + 180 mod 360 ReceivedAt time.Time `json:"received_at"` Raw string `json:"raw"` // Historical marks a spot recovered from a SH/DX table rather than heard live. // It belongs in the grid, but must NOT fire alerts or reach the panadapter: // replaying 100 past spots would spam both, and a station spotted three hours // ago is not on the air now. Historical bool `json:"historical,omitempty"` POTARef string `json:"pota_ref,omitempty"` // park id if this station is activating (api.pota.app) POTAName string `json:"pota_name,omitempty"` // park name } // State enumerates the per-server lifecycle. type State string const ( StateDisconnected State = "disconnected" StateConnecting State = "connecting" StateConnected State = "connected" StateReconnecting State = "reconnecting" StateError State = "error" ) // ServerStatus is one row of the runtime status table — one entry per // active session. type ServerStatus struct { ServerID int64 `json:"server_id"` Name string `json:"name"` Host string `json:"host"` Port int `json:"port"` State State `json:"state"` Login string `json:"login,omitempty"` Error string `json:"error,omitempty"` ConnectedAt time.Time `json:"connected_at,omitempty"` SpotsCount int `json:"spots_count,omitempty"` Retries int `json:"retries,omitempty"` } // session is one telnet connection bound to a single server config. // Line is one raw line of cluster traffic: everything the server says, plus the // commands we send (echoed, so the console reads as a conversation). // // Spots are PARSED OUT of this stream — but everything else used to be silently // dropped. You could type SH/DX/100 or WHO and never see the answer, which made // the command box look broken. The raw stream is the fix: the console shows what // the server actually said, spot or not. type Line struct { ServerID int64 `json:"server_id"` ServerName string `json:"server_name"` Text string `json:"text"` Sent bool `json:"sent"` // true = a command WE sent At string `json:"at"` // HH:MM:SS, local } // Internal — callers use Manager. The onStatus callback is fire-and- // forget: it tells the manager something changed; the frontend fetches // the new aggregate via Status() rather than receiving per-server diffs. type session struct { cfg ServerConfig login string onSpot func(Spot) onLine func(Line) onStatus func() mu sync.RWMutex status ServerStatus conn net.Conn stopCh chan struct{} doneCh chan struct{} stopped bool // guards against double-stop on the same session spotsCnt int dbgN int // diagnostic: how many raw lines logged this connection } // Manager owns N sessions, one per enabled server. Safe for concurrent // use from any goroutine; I/O is on per-session background goroutines. type Manager struct { mu sync.RWMutex sessions map[int64]*session onSpot func(Spot) onLine func(Line) onStatus func() } // NewManager builds an empty manager. emitSpot is called for each parsed // spot (with the source server filled in). emitStatusChanged is called // whenever ANY server's status changes — the frontend then re-fetches // the aggregate Status() via a Wails binding. func NewManager(emitSpot func(Spot), emitStatusChanged func(), emitLine func(Line)) *Manager { return &Manager{ sessions: make(map[int64]*session), onSpot: emitSpot, onLine: emitLine, onStatus: emitStatusChanged, } } // Status returns a snapshot of every running session's status. func (m *Manager) Status() []ServerStatus { m.mu.RLock() defer m.mu.RUnlock() out := make([]ServerStatus, 0, len(m.sessions)) for _, s := range m.sessions { out = append(out, s.snapshot()) } return out } // StartServer launches a session for cfg. login is the resolved callsign // to send (empty = anonymous). If a session for the same ID is already // running it is restarted with the new config. func (m *Manager) StartServer(cfg ServerConfig, login string) { m.StopServer(cfg.ID) if !cfg.Enabled || cfg.Host == "" { return } s := &session{ cfg: cfg, login: login, onSpot: m.onSpot, onLine: m.onLine, onStatus: m.emitStatus, stopCh: make(chan struct{}), doneCh: make(chan struct{}), status: ServerStatus{ ServerID: cfg.ID, Name: cfg.Name, Host: cfg.Host, Port: cfg.Port, Login: login, State: StateConnecting, }, } m.mu.Lock() m.sessions[cfg.ID] = s m.mu.Unlock() s.emitStatus() go s.run() } // StopServer terminates the session for the given ID (if any) and waits // for its goroutine to exit. func (m *Manager) StopServer(id int64) { m.mu.Lock() s, ok := m.sessions[id] if ok { delete(m.sessions, id) } m.mu.Unlock() if !ok { return } s.stop() m.emitStatus() } // StopAll closes every running session. func (m *Manager) StopAll() { m.mu.Lock() all := make([]*session, 0, len(m.sessions)) for _, s := range m.sessions { all = append(all, s) } m.sessions = make(map[int64]*session) m.mu.Unlock() for _, s := range all { s.stop() } m.emitStatus() } // SendCommand writes raw text (a CRLF is appended) to the session for // the given server ID. Used for "show last 30", "set/skimmer", etc. func (m *Manager) SendCommand(serverID int64, cmd string) error { m.mu.RLock() s, ok := m.sessions[serverID] m.mu.RUnlock() if !ok { return fmt.Errorf("no active session for server %d", serverID) } return s.send(cmd) } func (m *Manager) emitStatus() { if m.onStatus != nil { m.onStatus() } } // ---------- session ---------- func (s *session) snapshot() ServerStatus { s.mu.RLock() defer s.mu.RUnlock() return s.status } func (s *session) emitStatus() { if s.onStatus != nil { s.onStatus() } } func (s *session) send(cmd string) error { s.mu.RLock() conn := s.conn s.mu.RUnlock() if conn == nil { return fmt.Errorf("session %q not connected", s.cfg.Name) } _ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second)) _, err := conn.Write([]byte(strings.TrimRight(cmd, "\r\n") + "\r\n")) if err == nil { // Echo it into the console, so the operator sees WHAT was sent and can pair // it with the reply that follows. A console that only shows one side of the // conversation is barely better than none. s.emitLine(cmd, true) } return err } func (s *session) stop() { // Critical: do NOT nil out s.stopCh — the supervisor goroutine reads // `<-s.stopCh` in its select. Setting the field to nil would make // `<-nil` block forever, leaving the supervisor stuck on its backoff // timer and then re-dialing → a "deleted" cluster keeps spotting. // We just close() the channel and let the goroutine see the broadcast. s.mu.Lock() if s.stopped { s.mu.Unlock() return } s.stopped = true stop, done := s.stopCh, s.doneCh conn := s.conn s.mu.Unlock() if conn != nil { _ = conn.Close() } if stop != nil { close(stop) } if done != nil { <-done } } // run is the per-session supervisor: keeps trying to connect until // Stop is called. Backoff caps at 60s, resets after a 30s healthy link. func (s *session) run() { defer close(s.doneCh) backoff := []time.Duration{2, 5, 10, 30, 60} attempt := 0 for { select { case <-s.stopCh: return default: } connectedAt, err := s.runOnce() select { case <-s.stopCh: return default: } if !connectedAt.IsZero() && time.Since(connectedAt) > 30*time.Second { attempt = 0 } idx := attempt if idx >= len(backoff) { idx = len(backoff) - 1 } delay := backoff[idx] * time.Second attempt++ s.mu.Lock() s.status.State = StateReconnecting if err != nil { s.status.Error = err.Error() } s.status.Retries = attempt s.mu.Unlock() s.emitStatus() select { case <-s.stopCh: return case <-time.After(delay): } } } // runOnce dials, optionally logs in, sends init commands, parses spots. // Returns the moment we marked the link "connected" (zero if dial failed) // and the error that ended the session (nil if stopCh). func (s *session) runOnce() (time.Time, error) { addr := net.JoinHostPort(s.cfg.Host, fmt.Sprintf("%d", s.cfg.Port)) // IPv6-safe conn, err := net.DialTimeout("tcp", addr, 10*time.Second) if err != nil { return time.Time{}, fmt.Errorf("dial %s: %w", addr, err) } s.mu.Lock() s.conn = conn s.mu.Unlock() defer func() { s.mu.Lock() if s.conn == conn { s.conn = nil } s.mu.Unlock() _ = conn.Close() }() // Login: send on first prompt OR blindly after 1.5s. Many DXSpider // nodes accept the callsign without re-prompting. loginSent := false if s.login != "" { go func() { select { case <-s.stopCh: return case <-time.After(1500 * time.Millisecond): if !loginSent { _, _ = conn.Write([]byte(s.login + "\r\n")) } } }() } // Init commands, once per connection (so they replay after a reconnect). // // Timing: the cluster needs a moment after login before it will take commands, // hence the lead-in wait; then they are PACED, because a DXSpider/AR-Cluster // happily swallows a burst of back-to-back lines and silently ignores half of // them. A fast-but-dropped command is worse than a slow one. // // They go through s.send() — NOT a raw conn.Write, which is what they used to // do. That matters for two reasons: the raw write skipped the write deadline // (a wedged server could hang this goroutine forever), and it skipped the // console echo, so there was no way to SEE whether your init commands had // actually been applied. Now you watch them go out and the reply come back. initFired := false fireInitCommands := func() { if initFired || strings.TrimSpace(s.cfg.InitCommands) == "" { return } initFired = true go func() { time.Sleep(initCommandLeadIn) for _, line := range strings.Split(s.cfg.InitCommands, "\n") { line = strings.TrimSpace(strings.TrimRight(line, "\r")) if line == "" || strings.HasPrefix(line, "//") { continue } select { case <-s.stopCh: return default: } if err := s.send(line); err != nil { s.emitLine("init command failed: "+line+" — "+err.Error(), false) return } time.Sleep(initCommandGap) } }() } var connectedAt time.Time rd := bufio.NewReader(conn) for { select { case <-s.stopCh: return connectedAt, nil default: } _ = conn.SetReadDeadline(time.Now().Add(120 * time.Second)) line, err := rd.ReadString('\n') if err != nil { return connectedAt, fmt.Errorf("read: %w", err) } line = strings.TrimRight(line, "\r\n") // Strip terminal BELLs (\a) and other control bytes that some nodes // (e.g. F5LEN) append to spot lines — "DX de …0935Z KN04\a\a" — which // would otherwise break the parser's end-of-line anchor. Tabs → spaces. line = strings.Map(func(r rune) rune { if r == '\t' { return ' ' } if r < 0x20 || r == 0x7f { return -1 } return r }, line) if line == "" { continue } // Safety net: warn (rate-limited) about DX lines the parser still can't // read, so a future format drift is visible in the log. if strings.Contains(line, "DX de") && s.dbgN < 10 { if _, ok := parseSpot(line); !ok { s.dbgN++ applog.Printf("cluster[%s] UNPARSED spot: %q", s.cfg.Name, line) } } // Login on explicit prompt. if !loginSent && s.login != "" && isLoginPrompt(line) { _, _ = conn.Write([]byte(s.login + "\r\n")) loginSent = true continue } // Password on prompt (rare). if loginSent && s.cfg.Password != "" && isPasswordPrompt(line) { _, _ = conn.Write([]byte(s.cfg.Password + "\r\n")) continue } // Mark connected once we've sent login OR seen a welcome banner. if s.snapshot().State != StateConnected && (loginSent || isWelcome(line)) { connectedAt = time.Now() s.mu.Lock() s.status.State = StateConnected s.status.ConnectedAt = connectedAt s.status.Error = "" s.mu.Unlock() s.emitStatus() fireInitCommands() } // EVERY line goes to the console — spot or not. This is the whole point: // SH/DX, WHO, the MOTD and error replies are not spots, and dropping them // (which is what happened before) made the command box look inert. s.emitLine(line, false) // Live broadcast first; then the SH/DX table form. Without the second, a // SH/DX reply arrived, matched nothing, and vanished — which is exactly why // "SH/DX/100" looked like it did nothing at all. spot, ok := parseSpot(line) if !ok { spot, ok = parseShowDX(line) } if ok { spot.SourceID = s.cfg.ID spot.SourceName = s.cfg.Name s.mu.Lock() // Historical spots are a replay of the past, not new traffic — counting // them would inflate the server's "spots received" figure. if !spot.Historical { s.spotsCnt++ s.status.SpotsCount = s.spotsCnt } s.mu.Unlock() if s.onSpot != nil { s.onSpot(spot) } } } } // emitLine hands one line of traffic to the console. Blank lines are dropped — // clusters send plenty and they'd only pad the scrollback. func (s *session) emitLine(text string, sent bool) { if s.onLine == nil || strings.TrimSpace(text) == "" { return } s.onLine(Line{ ServerID: s.cfg.ID, ServerName: s.cfg.Name, Text: strings.TrimRight(text, "\r\n"), Sent: sent, At: time.Now().Format("15:04:05"), }) } // ---------- parsing ---------- // spotRE matches "DX de SPOTTER: FREQ DXCALL COMMENT TIME [LOC]". var spotRE = regexp.MustCompile( `^\s*DX\s+de\s+([A-Z0-9/#\-]+):?\s+(\d+\.?\d*)\s+([A-Z0-9/]+)\s+(.*?)\s+(\d{4}Z?)(?:\s+([A-R]{2}\d{2}(?:[A-X]{2})?))?\s*$`, ) // Pacing for the per-server init commands. // // A cluster is not ready to take commands the instant the login line goes out — // it is still printing its banner — so we wait before the first one. And they are // spaced, because DXSpider / AR-Cluster will happily accept a burst of // back-to-back lines and silently apply only some of them: the connection stays // up, no error is returned, your filters just aren't set. A command that is // dropped is far worse than one that is slow. const ( initCommandLeadIn = 1500 * time.Millisecond initCommandGap = 700 * time.Millisecond ) // showDXRE matches the reply to SH/DX — which is a TABLE, not the "DX de …" // broadcast format, and therefore matched nothing at all: // // 14195.0 EA8DHH 3-Jul-2026 1234Z CQ DX // freq dxcall date time comment // // This is why "SH/DX/100 does nothing": the 100 lines arrive, fail the broadcast // regex, and get dropped. The decimal point in the frequency is required — it is // what keeps ordinary prose out of the spot grid. var showDXRE = regexp.MustCompile( `^\s*(\d{3,7}\.\d+)\s+([A-Z0-9]{1,3}[0-9][A-Z0-9/]*)\s+(\d{1,2}-[A-Za-z]{3}-\d{4})\s+(\d{4})Z?\s*(.*?)\s*(?:<\s*([A-Z0-9/#\-]+)\s*>)?\s*$`, ) // parseShowDX turns one line of a SH/DX table into a Spot. The result is flagged // Historical: these are PAST spots, so they belong in the grid but must not fire // alerts or land on the panadapter — replaying 100 of them would spam both, and a // station spotted three hours ago is not on the air now. func parseShowDX(line string) (Spot, bool) { if strings.Contains(strings.ToUpper(line), "DX DE") { return Spot{}, false // that's the broadcast form; spotRE owns it } m := showDXRE.FindStringSubmatch(line) if m == nil { return Spot{}, false } khz, err := strconv.ParseFloat(m[1], 64) if err != nil || khz <= 0 { return Spot{}, false } hz := int64(khz * 1000) return Spot{ Spotter: strings.ToUpper(m[6]), DXCall: strings.ToUpper(m[2]), FreqKHz: khz, FreqHz: hz, Band: bandFromHz(hz), Comment: strings.TrimSpace(m[5]), TimeUTC: m[4] + "Z", ReceivedAt: time.Now(), Raw: line, Historical: true, }, true } func parseSpot(line string) (Spot, bool) { m := spotRE.FindStringSubmatch(line) if m == nil { return Spot{}, false } freqKHz, err := strconv.ParseFloat(m[2], 64) if err != nil { return Spot{}, false } freqHz := int64(freqKHz*1000 + 0.5) return Spot{ Spotter: strings.ToUpper(m[1]), FreqKHz: freqKHz, FreqHz: freqHz, Band: bandFromHz(freqHz), DXCall: strings.ToUpper(m[3]), Comment: strings.TrimSpace(m[4]), TimeUTC: m[5], Locator: strings.ToUpper(m[6]), ReceivedAt: time.Now(), Raw: line, }, true } func isLoginPrompt(s string) bool { low := strings.ToLower(s) return strings.Contains(low, "login:") || strings.Contains(low, "please enter your call") || strings.Contains(low, "your call:") || strings.HasSuffix(strings.TrimSpace(low), "callsign:") } func isPasswordPrompt(s string) bool { low := strings.ToLower(s) return strings.Contains(low, "password:") || strings.Contains(low, "pwd:") } func isWelcome(s string) bool { low := strings.ToLower(s) return strings.Contains(low, "welcome") || strings.Contains(low, "logged in") || strings.Contains(low, "started") } func bandFromHz(hz int64) string { mhz := float64(hz) / 1_000_000 switch { case mhz >= 1.8 && mhz <= 2.0: return "160m" case mhz >= 3.5 && mhz <= 4.0: return "80m" case mhz >= 5.3 && mhz <= 5.5: return "60m" case mhz >= 7.0 && mhz <= 7.3: return "40m" case mhz >= 10.1 && mhz <= 10.15: return "30m" case mhz >= 14.0 && mhz <= 14.35: return "20m" case mhz >= 18.068 && mhz <= 18.168: return "17m" case mhz >= 21.0 && mhz <= 21.45: return "15m" case mhz >= 24.89 && mhz <= 24.99: return "12m" case mhz >= 28.0 && mhz <= 29.7: return "10m" case mhz >= 50.0 && mhz <= 54.0: return "6m" case mhz >= 70.0 && mhz <= 70.5: return "4m" case mhz >= 144.0 && mhz <= 148.0: return "2m" case mhz >= 222.0 && mhz <= 225.0: return "1.25m" case mhz >= 420.0 && mhz <= 450.0: return "70cm" case mhz >= 902.0 && mhz <= 928.0: return "33cm" case mhz >= 1240.0 && mhz <= 1300.0: return "23cm" } return "" }