// svxdr — live-data daemon for SvxPortal (nodelist + proxy + astmeta + xlink) // Copyright (C) 2026 SA7AUX (Björn) // SPDX-License-Identifier: AGPL-3.0-or-later // // This program is free software: you can redistribute it and/or modify it under // the terms of the GNU Affero General Public License as published by the Free // Software Foundation, either version 3 of the License, or (at your option) any // later version. See the LICENSE file for details. // Package nodelist is the svxdr nodelist subsystem — the reflector status crawl, // lastheard, talker-meta API, external federation and xlink, folded in from the // standalone svx-web/nodelist. The logic is preserved verbatim; only config // wiring changed: env-var globals are overridden by applyConfig() from // svxdr.json, and main() became Run(ctx, cfg). See svxdrcfg.go. package nodelist import ( "bufio" "compress/gzip" "context" "crypto/sha256" "crypto/subtle" "crypto/tls" "encoding/hex" "encoding/json" "fmt" "io" "log" "net" "net/http" "net/url" "os" "path/filepath" "regexp" "sort" "strconv" "strings" "sync" "time" mqtt "github.com/eclipse/paho.mqtt.golang" "github.com/redis/go-redis/v9" "svxportal/svxdr/config" ) var ( ctx = context.Background() listenAddr = env("LISTEN_ADDR", ":8098") reflectorURL = env("REFLECTOR_STATUS_URL", "http://127.0.0.1:8090/status") pollInterval = envDur("POLL_INTERVAL", 1*time.Second) redisSock = env("REDIS_SOCKET", "/run/redis/redis.sock") redisPass = env("REDIS_PASSWORD", "") statusKey = env("REDIS_STATUS_KEY", "/registry/svxlink/status/json") statusHashKey = env("REDIS_STATUS_HASH_KEY", "/registry/svxlink/status/hash") pubsubChan = env("REDIS_PUBSUB_CHAN", "/registry/svxlink/status/update") nodesOnlineKey = env("REDIS_NODES_ONLINE_KEY", "/registry/svxlink/nodes/online") nodeKeyPrefix = env("REDIS_NODE_KEY_PREFIX", "/registry/svxlink/node") nodeConnectChan = env("REDIS_NODE_CONNECT_CHAN", "/registry/svxlink/node/connect") nodeDisconnectChan = env("REDIS_NODE_DISCONNECT_CHAN", "/registry/svxlink/node/disconnect") nodeUpdateChan = env("REDIS_NODE_UPDATE_CHAN", "/registry/svxlink/node/update") lastheardKey = env("REDIS_LASTHEARD_KEY", "/registry/svxlink/lastheard") lastheardWindowSec = int64(24 * 60 * 60) reflectorLogPath = env("REFLECTOR_LOG_PATH", "/var/log/svxreflector") nodeOnlineTTL = envDur("NODE_ONLINE_TTL", 0) nodeTTL = envDur("NODE_TTL", 0) // How long a talker-meta entry (POST /api/talker-meta) stays valid // without refresh (astmeta re-injects USRP INFO every ~10s while the // talker is keyed, which re-fires the TCL hook that posts here). talkerMetaTTL = envDur("TALKER_META_TTL", 30*time.Second) // Shared secret for POST /api/talker-meta (the svxlink TCL hook on the // USRP bridge). Empty = endpoint disabled. ingestToken = env("TALKER_INGEST_TOKEN", "") // How long a bridge-heard station stays in the /api/allstar-nodes list // after its last keyup. allstarNodesTTL = envDur("ALLSTAR_NODES_TTL", 30*time.Minute) // MQTT (reflector publishes the full picture incl. trunk-side nodes here). mqttBroker = env("MQTT_BROKER", "") // e.g. tcp://127.0.0.1:1883 mqttUser = env("MQTT_USERNAME", "") mqttPass = env("MQTT_PASSWORD", "") mqttPrefix = env("MQTT_TOPIC_PREFIX", "") // e.g. svxreflector/sa7aux // Cross-link talker mesh (SVX side). Publishes local SVX talkers to the mesh // and injects DMR/AllStar talkers into the SVX lastheard. The DMR side owns // the TG maps + translation, so this side only ever deals in its own svx_tg. xlinkEnabled = envBool("XLINK_ENABLED", false) xlinkBroker = env("XLINK_BROKER", "") // tcp://host:port xlinkUser = env("XLINK_USER", "") xlinkPass = env("XLINK_PASS", "") xlinkTopic = env("XLINK_TOPIC", "callmgr/xlink/talker") xlinkOrigin = env("XLINK_ORIGIN", "") // defaults to svx@ // External network pull: upstream aggregate of every reflector in the // network (sources + their nodes tagged _source + all network trunks). // The uplink host is derived from svxreflector.conf at runtime — NO address // is baked into the binary. Scheme/port/path are env-tunable because the // topology HTTP port varies per system/trunk; port is never assumed. svxreflectorConf = env("SVXREFLECTOR_CONF", "/etc/svxlink/svxreflector.conf") // chan_svx_sat config: maps each ENABLED satellite trunk's satellite_id // (e.g. sat403872) to its friendly callsign (e.g. KT4ROY-SAT1). Disabled // (commented) sections are absent, so only enabled sat trunks are named/shown. svxSatConf = env("SVX_SAT_CONF", "/etc/asterisk/svx_sat.conf") externalPullURL = env("EXTERNAL_PULL_URL", "") // optional full override; else derived from conf externalPullScheme = env("EXTERNAL_PULL_SCHEME", "https") externalPullPort = env("EXTERNAL_PULL_PORT", "") // topology (svx_externalpull.php): empty = scheme default (443/80) externalPullPath = env("EXTERNAL_PULL_PATH", "/svx_externalpull.php") externalPullInterval = envDur("EXTERNAL_PULL_INTERVAL", 60*time.Second) // gentle: spread crawl, ~1/min // Per-ref /status crawl. Each reflector's own /status gives its nodes AND its // trunks, whose hosts branch out to more reflectors — so we crawl the graph. externalStatusPorts = env("EXTERNAL_STATUS_PORTS", "") // per-server /status ports (comma array) to try beyond 443/80 — spec in /etc/default/nodelist; NO port baked in (svxref is often 8080 but owners vary, e.g. 8090) externalStatusPath = env("EXTERNAL_STATUS_PATH", "/status") externalProxyPath = env("EXTERNAL_PROXY_PATH", "/proxy_status.php") // dashboard's real per-ref status (nodes+trunks+trunks_UDP+routing) externalDetailPath = env("EXTERNAL_DETAIL_PATH", "/geu/status/") // merged roster with full per-node detail (location/freq/sw), enriches route nodes externalMaxRefs = envInt("EXTERNAL_MAX_REFS", 100) // crawl bound (cycle-safe), 0 = unlimited externalStaleTTL = envDur("EXTERNAL_STALE_TTL", 5*time.Minute) // keep a reflector's last-good trunks this long when its probe times out — smooths flicker; 0 disables retention // Federation seeds: reflector /status URLs that are NOT reachable via our own // trunk topology — they don't advertise themselves in the mesh, so the crawl // never finds them on its own (e.g. the Swedish Reflector-network core). Seed // them explicitly; the recursive crawl then reads each one's nodes and follows // its trunks like any other reflector. Comma-separated full /status URLs, // set in the site config (FEDERATION_SEEDS env / svxdr.json federation.seeds). // Empty = no seeding. There is deliberately NO built-in list: seeds are // site-specific runtime config and must never be hardcoded here. federationSeeds = env("FEDERATION_SEEDS", "") // no built-in seeds: configure per site, empty = no seeding // Fas 2: federation gap detector. Periodically fetch the network-wide // aggregate (sa2blv's reflector-network) and diff it against what our own // recursive crawl covers, so we can see which reflectors/nodes we are missing // and add them to FEDERATION_SEEDS. Read-only — never merged into the roster // (our own fresh crawl/MQTT data stays the single source of truth). Empty URL // disables the detector. federationAggURL = env("FEDERATION_AGGREGATE_URL", "") // site config; empty disables the detector federationAggInterval = envDur("FEDERATION_AGGREGATE_INTERVAL", 5*time.Minute) debug = envBool("DEBUG", false) ) // mqttState holds the latest node lists published per source on MQTT // (/nodes/local and /nodes/). Keyed src -> callsign -> raw. var mqttState = struct { mu sync.RWMutex bySrc map[string]map[string]json.RawMessage }{bySrc: map[string]map[string]json.RawMessage{}} // mergeMqttNodes returns local nodes (authoritative) overlaid on every MQTT // source's nodes, so trunk-side nodes (SE-EASTCOAST, SM-RO, …) appear too. func mergeMqttNodes(local map[string]json.RawMessage) map[string]json.RawMessage { merged := map[string]json.RawMessage{} mqttState.mu.RLock() for src, m := range mqttState.bySrc { if src == "local" { continue // HTTP /status is authoritative for local nodes } for cs, raw := range m { merged[cs] = raw } } mqttState.mu.RUnlock() for cs, raw := range local { merged[cs] = raw } return merged } // onNodesMsg handles a /nodes/ message: {"nodes":[{callsign,...},...]}. func onNodesMsg(topic string, payload []byte) { src := topic[strings.LastIndex(topic, "/")+1:] var p struct { Nodes []json.RawMessage `json:"nodes"` } if err := json.Unmarshal(payload, &p); err != nil { return } m := map[string]json.RawMessage{} for _, raw := range p.Nodes { var h struct { Callsign string `json:"callsign"` } if json.Unmarshal(raw, &h) != nil || h.Callsign == "" { continue } stored := raw // Tag trunk-sourced nodes so the UI can tell them apart from local // clients (only local clients can be blocked via NODE BLOCK). if src != "local" { var obj map[string]json.RawMessage if json.Unmarshal(raw, &obj) == nil { if sj, err := json.Marshal(src); err == nil { obj["remoteTrunk"] = sj if nb, err := json.Marshal(obj); err == nil { stored = nb } } } } m[h.Callsign] = stored } mqttState.mu.Lock() mqttState.bySrc[src] = m mqttState.mu.Unlock() if debug { log.Printf("DEBUG mqtt nodes src=%s count=%d", src, len(m)) } } func startMqtt() { defer func() { if r := recover(); r != nil { log.Printf("mqtt goroutine panic: %v", r) } }() topic := strings.TrimRight(mqttPrefix, "/") + "/nodes/+" opts := mqtt.NewClientOptions(). AddBroker(mqttBroker). SetClientID("nodelist"). SetUsername(mqttUser). SetPassword(mqttPass). SetAutoReconnect(true). SetConnectRetry(true). SetConnectRetryInterval(5 * time.Second). SetOnConnectHandler(func(c mqtt.Client) { if t := c.Subscribe(topic, 0, func(_ mqtt.Client, msg mqtt.Message) { onNodesMsg(msg.Topic(), msg.Payload()) }); t.Wait() && t.Error() != nil { log.Printf("mqtt subscribe failed: %v", t.Error()) } else { log.Printf("mqtt subscribed: %s", topic) } }). SetConnectionLostHandler(func(_ mqtt.Client, err error) { log.Printf("mqtt connection lost: %v", err) }) c := mqtt.NewClient(opts) if t := c.Connect(); t.Wait() && t.Error() != nil { log.Printf("mqtt connect failed: %v (broker=%s)", t.Error(), mqttBroker) } } // ---------------------------------------------------------- cross-link mesh -- // xlinkMsg is one talker announcement on the cross-link topic. tg is always in // the RECEIVER's numbering (the DMR side translates before publishing), so this // side only ever deals in its own svx_tg. type xlinkMsg struct { Origin string `json:"origin"` Network string `json:"network"` // dmr | svx | allstar (the SIDE, for routing/loop-guard) Callsign string `json:"callsign"` System string `json:"system"` // system/reflector name (e.g. East Coast Reflector), distinct from callsign Trunk string `json:"trunk"` // trunk/link the talker enters through (e.g. KT4ROY-SAT1) → "callsign via trunk" Node string `json:"node"` // source node/id (e.g. AllStar node 45225) → DMR src_id TG int `json:"tg"` Backend string `json:"backend"` // display backend: AllStar / SVXLink / trunk … Event string `json:"event"` // start | end TS int64 `json:"ts"` } var xlinkClient mqtt.Client // xlinkStart connects the SVX side to the cross-link mesh (if enabled) and // subscribes for peer talkers to inject into the SVX lastheard. func xlinkStart(rdb *redis.Client) { if !xlinkEnabled || xlinkBroker == "" { return } if xlinkOrigin == "" { h, _ := os.Hostname() xlinkOrigin = "svx@" + h } opts := mqtt.NewClientOptions(). AddBroker(xlinkBroker). SetClientID(fmt.Sprintf("nodelist-xlink-%d", time.Now().UnixNano())). SetUsername(xlinkUser). SetPassword(xlinkPass). SetAutoReconnect(true). SetConnectRetry(true). SetConnectRetryInterval(5 * time.Second). SetOnConnectHandler(func(c mqtt.Client) { if t := c.Subscribe(xlinkTopic, 0, func(_ mqtt.Client, m mqtt.Message) { xlinkOnMessage(rdb, m.Payload()) }); t.Wait() && t.Error() != nil { log.Printf("xlink subscribe failed: %v", t.Error()) } else { log.Printf("xlink subscribed: %s", xlinkTopic) } }) xlinkClient = mqtt.NewClient(opts) xlinkClient.Connect() log.Printf("xlink: mesh enabled broker=%s topic=%s origin=%s", xlinkBroker, xlinkTopic, xlinkOrigin) } // xlinkPublishTalker announces a local SVX talker to the mesh (own numbering). // The DMR side gates on its TG map, so publishing every talker is harmless. func xlinkPublishTalker(callsign, system, trunk, node string, tg int, backend, event string, ts int64) { if xlinkClient == nil || tg <= 0 { return } if backend == "" { backend = "SVXLink" } b, _ := json.Marshal(xlinkMsg{Origin: xlinkOrigin, Network: "svx", Callsign: callsign, System: system, Trunk: trunk, Node: node, TG: tg, Backend: backend, Event: event, TS: ts}) xlinkClient.Publish(xlinkTopic, 0, false, b) } // xlinkOnMessage injects a peer (DMR/AllStar) talker into the SVX lastheard. // Own-origin and svx-origin messages are ignored (loop guard); injected rows go // straight to Redis (not the reflector log), so they are never re-published. func xlinkOnMessage(rdb *redis.Client, payload []byte) { var m xlinkMsg if json.Unmarshal(payload, &m) != nil { return } if m.Origin == xlinkOrigin || m.Network == "svx" || m.Network == "" { return } if m.Event != "start" || m.TG <= 0 { return // show on start; the DMR side is authoritative on gating/TG } ts := m.TS if ts == 0 { ts = time.Now().Unix() } backend := m.Backend if backend == "" { backend = "DMR" } writeLastheardEvent(rdb, lastheardEvent{ Ts: ts, Callsign: m.Callsign, TG: m.TG, Duration: 1, Source: "crosslink", Network: backend, // Location = the TG's reflector name (e.g. 314093 -> "WHITELIGHTNING"), so a // cross-linked DMR talker gets the same Location label as an AllStar talker // on that TG. From nodes[].system (svx_sat.conf), via locForTG's fallback. Location: locForTG(m.TG), }) } // extTrunkRef is one trunk as owned by a specific server: its name, the host on // the other end, and whether it's up. Keeping ownership lets the topology graph // draw the real server->server links instead of hanging everything on center. type extTrunkRef struct { Name string `json:"name"` Host string `json:"host"` Connected bool `json:"connected"` Kind string `json:"kind"` // "trunk" | "udp" | "satellite" } // extSource is one reflector (server) in the network: the nodes connected to it // and the trunks it owns. IP/HTTPPort/Aliases feed the admin info box. type extSource struct { Tag string `json:"tag"` URL string `json:"url"` OK bool `json:"ok"` Version string `json:"version,omitempty"` Mode string `json:"mode,omitempty"` IP string `json:"ip,omitempty"` HTTPPort string `json:"http_port,omitempty"` Aliases []string `json:"aliases,omitempty"` // other hostnames resolving to the same IP Nodes map[string]json.RawMessage `json:"nodes"` Trunks []extTrunkRef `json:"trunks"` Routing []routeEntry `json:"-"` // this ref's Routing_table (callsign -> trunk IP), for node attribution Detail map[string]json.RawMessage `json:"-"` // merged-roster per-node detail, to enrich route nodes } // externalState holds the latest network-wide picture pulled from // EXTERNAL_PULL_URL: sources (each with its grouped nodes, sorted by tag) and // the raw trunks map covering the whole network. var externalState = struct { mu sync.RWMutex sources []extSource // sorted by Tag trunks json.RawMessage // raw {name: trunkobj} map routing map[string]string // trunk IP -> callsign (from reflectors' Routing_table) generated string lastOKAt time.Time lastErr string // Network-wide bridge directory: every crawled reflector's OWN bridges // (its /api/bridge-nodes), unioned. Lets any box show "connected to // (node)" for a bridge that lives on another server, derived // purely from the topology it already crawls — no per-box config. bridgeNodes map[string]bridgeInfo // UPPER(via) -> {node, system} }{} // pullSource is one reflector as listed in the topology (svx_externalpull.php). // Its URL is the owner-published status endpoint we read directly for nodes. type pullSource struct { Tag string `json:"tag"` URL string `json:"url"` OK bool `json:"ok"` Version string `json:"version"` Mode string `json:"mode"` } // externalPull mirrors the fields we consume from svx_externalpull.php. It is // used purely as the TOPOLOGY (the list of reflectors + the network trunk list); // per-ref nodes are read directly from each reflector's own /status, not taken // from this aggregate. type externalPull struct { Generated string `json:"generated"` Sources []pullSource `json:"sources"` Trunks json.RawMessage `json:"trunks"` } // uplinkHostsFromConf parses svxreflector.conf and returns the trunk-peer hosts // that form this reflector's uplink(s): // // [ReflectorTrunk] Peers=SE-AUX,... -> [TrunkPeer#SE-AUX] Host= // // It also handles the twin2 form where each trunk is its own [TRUNK_*] section // with the peer HOST inline. No address is baked into the binary; the uplink // follows the config. Note the peer's Port= is the trunk protocol port // (5500/5302), NOT the HTTP topology port, so it is deliberately ignored here. func uplinkHostsFromConf(path string) []string { f, err := os.Open(path) if err != nil { return nil } defer f.Close() var peers []string hostByPeer := map[string]string{} var directHosts []string // twin2 [TRUNK_*] sections carry the peer host inline curPeer := "" // current [TrunkPeer#] section inTrunk := false // inside [ReflectorTrunk] inTrunkSec := false // inside a twin2 [TRUNK_*] section sc := bufio.NewScanner(f) for sc.Scan() { line := strings.TrimSpace(sc.Text()) if line == "" || strings.HasPrefix(line, "#") { continue } if strings.HasPrefix(line, "[") && strings.HasSuffix(line, "]") { sec := line[1 : len(line)-1] inTrunk = strings.EqualFold(sec, "ReflectorTrunk") inTrunkSec = strings.HasPrefix(strings.ToUpper(sec), "TRUNK_") curPeer = "" if i := strings.IndexByte(sec, '#'); i >= 0 && strings.EqualFold(sec[:i], "TrunkPeer") { curPeer = sec[i+1:] } continue } eq := strings.IndexByte(line, '=') if eq < 0 { continue } key := strings.TrimSpace(line[:eq]) val := strings.TrimSpace(line[eq+1:]) switch { case inTrunk && strings.EqualFold(key, "Peers"): for _, p := range strings.Split(val, ",") { if p = strings.TrimSpace(p); p != "" { peers = append(peers, p) } } case curPeer != "" && strings.EqualFold(key, "Host"): hostByPeer[curPeer] = val case inTrunkSec && strings.EqualFold(key, "Host"): directHosts = append(directHosts, val) } } var hosts []string seen := map[string]struct{}{} add := func(h string) { if h == "" { return } if _, ok := seen[h]; ok { return } seen[h] = struct{}{} hosts = append(hosts, h) } for _, p := range peers { add(hostByPeer[p]) } for _, h := range directHosts { add(h) } return hosts } // satTGsFromConf parses chan_svx_sat's config (svx_sat.conf) and groups every // ENABLED entry by its satellite_id, returning the talkgroups (per-TG callsign + // tg) each satellite carries. Several entries sharing a satellite_id are the SAME // trunk with several TGs, so they are grouped — not collapsed to the last callsign. func satTGsFromConf(path string) map[string][]satTG { out := map[string][]satTG{} f, err := os.Open(path) if err != nil { return out } defer f.Close() var satID, call, tg string flush := func() { if satID != "" { out[satID] = append(out[satID], satTG{Name: call, TG: tg}) } satID, call, tg = "", "", "" } sc := bufio.NewScanner(f) for sc.Scan() { line := strings.TrimSpace(sc.Text()) if line == "" || strings.HasPrefix(line, ";") || strings.HasPrefix(line, "#") { continue } if strings.HasPrefix(line, "[") && strings.HasSuffix(line, "]") { flush() continue } eq := strings.IndexByte(line, '=') if eq < 0 { continue } switch strings.ToLower(strings.TrimSpace(line[:eq])) { case "satellite_id": satID = strings.TrimSpace(line[eq+1:]) case "callsign": call = strings.TrimSpace(line[eq+1:]) case "tg": tg = strings.TrimSpace(line[eq+1:]) } } flush() return out } // satelliteIDsFromReflectorConf parses svxreflector.conf's [SATELLITE] section and // returns the configured satellite ids (the in each SECRET_=… line). Only // the id is read — the secret value is never touched or exposed. Lets the Trunks // view list every configured satellite (online or offline) fully dynamically. func satelliteIDsFromReflectorConf(path string) []string { f, err := os.Open(path) if err != nil { return nil } defer f.Close() var ids []string seen := map[string]bool{} inSat := false sc := bufio.NewScanner(f) for sc.Scan() { line := strings.TrimSpace(sc.Text()) if line == "" || strings.HasPrefix(line, "#") || strings.HasPrefix(line, ";") { continue } if strings.HasPrefix(line, "[") && strings.HasSuffix(line, "]") { inSat = strings.EqualFold(line, "[SATELLITE]") continue } if !inSat { continue } eq := strings.IndexByte(line, '=') if eq < 0 { continue } key := strings.TrimSpace(line[:eq]) if len(key) > 7 && strings.EqualFold(key[:7], "SECRET_") { if id := key[7:]; id != "" && !seen[id] { seen[id] = true ids = append(ids, id) } } } return ids } // satNamesFromConf parses chan_svx_sat's config (svx_sat.conf) and returns a map // from each ENABLED satellite trunk's satellite_id (e.g. "sat403872") to its // friendly callsign (e.g. "KT4ROY-SAT1"). Commented-out (disabled) sections and // keys are skipped, so only enabled sat trunks are surfaced. Empty if unreadable. func satNamesFromConf(path string) map[string]string { out := map[string]string{} f, err := os.Open(path) if err != nil { return out } defer f.Close() var satID, call string flush := func() { if satID != "" && call != "" { out[satID] = call } satID, call = "", "" } sc := bufio.NewScanner(f) for sc.Scan() { line := strings.TrimSpace(sc.Text()) if line == "" || strings.HasPrefix(line, ";") || strings.HasPrefix(line, "#") { continue // blank or disabled/commented → skip } if strings.HasPrefix(line, "[") && strings.HasSuffix(line, "]") { flush() // section boundary — commit the previous section continue } eq := strings.IndexByte(line, '=') if eq < 0 { continue } switch strings.ToLower(strings.TrimSpace(line[:eq])) { case "satellite_id": satID = strings.TrimSpace(line[eq+1:]) case "callsign": call = strings.TrimSpace(line[eq+1:]) } } flush() return out } // externalPullURLs returns the topology URL(s) to fetch. EXTERNAL_PULL_URL wins // if set; otherwise one URL per uplink host derived from the conf. Neither host // nor port is ever baked in — port comes from EXTERNAL_PULL_PORT (empty = scheme // default) because it varies per system/trunk. func externalPullURLs() []string { if externalPullURL != "" { return []string{externalPullURL} } hosts := uplinkHostsFromConf(svxreflectorConf) scheme := externalPullScheme if scheme == "" { scheme = "https" } path := externalPullPath if !strings.HasPrefix(path, "/") { path = "/" + path } urls := make([]string, 0, len(hosts)) for _, h := range hosts { host := h if externalPullPort != "" { host = net.JoinHostPort(h, externalPullPort) } urls = append(urls, fmt.Sprintf("%s://%s%s", scheme, host, path)) } return urls } // fetchExternalPull GETs and parses one topology endpoint (svx_externalpull.php). func fetchExternalPull(client *http.Client, u string) (externalPull, error) { var p externalPull req, err := http.NewRequest(http.MethodGet, u, nil) if err != nil { return p, err } req.Header.Set("Accept", "application/json") resp, err := client.Do(req) if err != nil { return p, err } defer resp.Body.Close() if resp.StatusCode != 200 { _, _ = io.Copy(io.Discard, resp.Body) return p, fmt.Errorf("http %d", resp.StatusCode) } body, err := io.ReadAll(resp.Body) if err != nil { return p, err } if err := json.Unmarshal(body, &p); err != nil { return p, fmt.Errorf("parse: %w", err) } return p, nil } // hostFromURL extracts the host (no port) from a URL, "" on failure. func hostFromURL(raw string) string { if raw == "" { return "" } u, err := url.Parse(raw) if err != nil { return "" } return u.Hostname() } // normalizeHost lowercases and strips scheme/port for dedupe keying. func normalizeHost(h string) string { h = strings.TrimSpace(strings.ToLower(h)) if h == "" { return "" } if strings.Contains(h, "://") { if hh := hostFromURL(h); hh != "" { return strings.ToLower(hh) } } if host, _, err := net.SplitHostPort(h); err == nil { return host } return h } // resolveCache memoises hostname -> primary IP so the same host isn't looked up // twice within a crawl. var ( resolveMu sync.Mutex resolveCache = map[string]string{} ) // refCacheEntry is one reflector's last SUCCESSFUL crawl contribution, retained so // a single timed-out probe (or a briefly-unreachable subtree) doesn't erase its // trunks. Keyed by hostKey — the same dedup key the crawl uses for `visited`. type refCacheEntry struct { src extSource // the reflector's extSource incl. its owned trunk refs trunks map[string]json.RawMessage // raw trunk/udp/satellite blobs it contributed (name -> raw) routing []routeEntry // its Routing_table entries (trunk IP -> callsign) at time.Time // when this reflector was last read OK } // refCache holds last-known-good per reflector so crawlNetwork can resurrect a // reflector whose probe failed this cycle, until it's been unreachable for // externalStaleTTL. Touched only in the single-threaded fold, never by probe // goroutines; the mutex is belt-and-suspenders. var ( refCacheMu sync.Mutex refCache = map[string]refCacheEntry{} ) // putRefCache stores a reflector's just-read good contribution, stamped now. func putRefCache(key string, e refCacheEntry) { e.at = time.Now() refCacheMu.Lock() refCache[key] = e refCacheMu.Unlock() } // freshRefCache returns a reflector's retained contribution if present and read // within externalStaleTTL; stale entries are pruned so the map stays bounded. func freshRefCache(key string) (refCacheEntry, bool) { if key == "" || externalStaleTTL <= 0 { return refCacheEntry{}, false } refCacheMu.Lock() defer refCacheMu.Unlock() e, ok := refCache[key] if !ok { return refCacheEntry{}, false } if time.Since(e.at) > externalStaleTTL { delete(refCache, key) return refCacheEntry{}, false } return e, true } // hostKey returns a dedup key for a host: its resolved IP when possible (so two // hostnames for the same physical reflector — e.g. sm2ampr.net and // reflector-network.se both 44.5.25.40 — collapse to one), else the bare host. func hostKey(host string) string { h := normalizeHost(host) if h == "" { return "" } resolveMu.Lock() if v, ok := resolveCache[h]; ok { resolveMu.Unlock() return v } resolveMu.Unlock() key := h if ips, err := net.LookupHost(h); err == nil && len(ips) > 0 { sort.Strings(ips) key = ips[0] } resolveMu.Lock() resolveCache[h] = key resolveMu.Unlock() return key } // isIPStr reports whether h is a bare IP literal. func isIPStr(h string) bool { return net.ParseIP(strings.Trim(h, "[]")) != nil } // trunkHost reads the "host" field from a trunk status object. func trunkHost(raw json.RawMessage) string { var t struct { Host string `json:"host"` } if json.Unmarshal(raw, &t) != nil { return "" } return strings.TrimSpace(t.Host) } // trunkConnected reads the "connected" flag from a trunk object. func trunkConnected(raw json.RawMessage) bool { var t struct { Connected bool `json:"connected"` } _ = json.Unmarshal(raw, &t) return t.Connected } func ensureSlash(p string) string { if !strings.HasPrefix(p, "/") { return "/" + p } return p } // statusCandidates returns the ordered status URLs to try for a reflector. The // dashboard's proxy_status.php is tried FIRST — it exposes the real, full status // (nodes + trunks + trunks_UDP + Routing_table); a raw /status can be empty on // hub reflectors. The HTTP port varies per owner and isn't published, so we try // 443 (https) and 80 (http) — common behind a proxy — plus each configured extra // port (EXTERNAL_STATUS_PORTS, spec'd per server — no port assumed), with proxy + /status. func statusCandidates(host, publishedURL string, ports []string) []string { h := host if h == "" { h = hostFromURL(publishedURL) } proxy := ensureSlash(externalProxyPath) stat := ensureSlash(externalStatusPath) var c []string if h != "" { c = append(c, "https://"+h+proxy) // proxy_status.php over 443 first } if publishedURL != "" { c = append(c, publishedURL) } if h != "" { // 443 (https) and 80 (http) — common behind nginx/apache c = append(c, "https://"+h+stat, "http://"+h+proxy, "http://"+h+stat) // ports learned from the network (each svxref's http_port) + configured seeds for _, p := range ports { p = strings.TrimSpace(p) if p == "" { continue } hp := net.JoinHostPort(h, p) c = append(c, "http://"+hp+proxy, "http://"+hp+stat) } } return c } // refStatus is one reflector's own status: its nodes, its named trunks, its UDP // trunk peers, and the HTTP port it reports for itself (http_port) — the crawl // learns real ports from the network instead of guessing. type routeEntry struct { Callsign string Trunk string // the trunk IP this callsign is reachable via TG int MonitoredTGs []int } type refStatus struct { Nodes map[string]json.RawMessage Trunks map[string]json.RawMessage TrunksUDP []json.RawMessage Satellites map[string]json.RawMessage // id -> full satellite obj (incl. its nodes[]) HTTPPort string Version string Mode string Routing []routeEntry Detail map[string]json.RawMessage // full per-node detail from the merged roster (enriches route nodes) } // jsonToPort reads http_port whether emitted as a string ("8085") or a number. func jsonToPort(raw json.RawMessage) string { var s string if json.Unmarshal(raw, &s) == nil { return strings.TrimSpace(s) } var n json.Number if json.Unmarshal(raw, &n) == nil { return n.String() } return "" } // parseNodes accepts a reflector's "nodes" either as an object keyed by callsign // or as an array of node objects (different reflectors/versions emit both), and // returns a callsign-keyed map. func parseNodes(raw json.RawMessage) map[string]json.RawMessage { var m map[string]json.RawMessage if json.Unmarshal(raw, &m) == nil { return m } var arr []json.RawMessage if json.Unmarshal(raw, &arr) == nil { out := make(map[string]json.RawMessage, len(arr)) for _, r := range arr { var h struct { Callsign string `json:"callsign"` } if json.Unmarshal(r, &h) == nil && h.Callsign != "" { out[h.Callsign] = r } } return out } return map[string]json.RawMessage{} } // parseRefStatus parses a /status body into nodes + trunks. Requires a top-level // "nodes" key so a stray 200 (HTML/redirect page) is rejected. func parseRefStatus(body []byte) (refStatus, bool) { var top map[string]json.RawMessage if json.Unmarshal(body, &top) != nil { return refStatus{}, false } if _, ok := top["nodes"]; !ok { return refStatus{}, false } rs := refStatus{Nodes: parseNodes(top["nodes"]), Trunks: map[string]json.RawMessage{}} if raw, ok := top["trunks"]; ok { var tr map[string]json.RawMessage if json.Unmarshal(raw, &tr) == nil { rs.Trunks = tr } } if raw, ok := top["trunks_UDP"]; ok { var arr []json.RawMessage if json.Unmarshal(raw, &arr) == nil { rs.TrunksUDP = arr } } if raw, ok := top["satellites"]; ok { var m map[string]json.RawMessage if json.Unmarshal(raw, &m) == nil { rs.Satellites = m } } if raw, ok := top["http_port"]; ok { rs.HTTPPort = jsonToPort(raw) } if raw, ok := top["version"]; ok { _ = json.Unmarshal(raw, &rs.Version) } if raw, ok := top["mode"]; ok { _ = json.Unmarshal(raw, &rs.Mode) } if raw, ok := top["Routing_table"]; ok { var arr []struct { Callsign string `json:"callsign"` Trunk string `json:"trunk"` TG int `json:"tg"` TGMonitor []int `json:"tg_monitor"` } if json.Unmarshal(raw, &arr) == nil { for _, e := range arr { if e.Callsign != "" && e.Trunk != "" { rs.Routing = append(rs.Routing, routeEntry{ Callsign: e.Callsign, Trunk: strings.TrimSpace(e.Trunk), TG: e.TG, MonitoredTGs: e.TGMonitor, }) } } } } return rs, true } // normalizeUDPTrunk turns a trunks_UDP entry into a uniform trunk object: it // maps status->connected so the same renderer works for named + UDP trunks, and // returns the peer name and host for dedup/branching. func normalizeUDPTrunk(raw json.RawMessage) (name string, out json.RawMessage, host string) { var m map[string]json.RawMessage if json.Unmarshal(raw, &m) != nil { return "", raw, "" } if b, ok := m["name"]; ok { _ = json.Unmarshal(b, &name) } if b, ok := m["host"]; ok { _ = json.Unmarshal(b, &host) } if st, ok := m["status"]; ok { if _, has := m["connected"]; !has { m["connected"] = st } } if b, err := json.Marshal(m); err == nil { out = b } else { out = raw } return name, out, strings.TrimSpace(host) } // readRefStatus tries each candidate URL until one returns a parseable /status. // fetchDetailRoster fetches the reflector's MERGED node roster (full per-node // detail: location, freqs, sw, sysop) from externalDetailPath derived off the // working status URL (same scheme+host, different path). proxy_status.php only // lists a UDP-peer member's callsign in Routing_table, so this fills in detail. // Best-effort; returns nil on any failure (route nodes then stay sparse). func fetchDetailRoster(client *http.Client, usedURL string) map[string]json.RawMessage { if externalDetailPath == "" || usedURL == "" { return nil } u, err := url.Parse(usedURL) if err != nil { return nil } u.Path = ensureSlash(externalDetailPath) u.RawQuery = "" req, err := http.NewRequest(http.MethodGet, u.String(), nil) if err != nil { return nil } req.Header.Set("Accept", "application/json") resp, err := client.Do(req) if err != nil { return nil } body, rerr := io.ReadAll(io.LimitReader(resp.Body, 16<<20)) resp.Body.Close() if rerr != nil || resp.StatusCode != 200 { return nil } var top map[string]json.RawMessage if json.Unmarshal(body, &top) != nil { return nil } nm := parseNodes(top["nodes"]) if len(nm) == 0 { return nil } return nm } func readRefStatus(client *http.Client, host, publishedURL string, ports []string) (rs refStatus, usedURL string, ok bool) { for _, u := range statusCandidates(host, publishedURL, ports) { req, err := http.NewRequest(http.MethodGet, u, nil) if err != nil { continue } req.Header.Set("Accept", "application/json") resp, err := client.Do(req) if err != nil { continue } body, rerr := io.ReadAll(io.LimitReader(resp.Body, 16<<20)) resp.Body.Close() if rerr != nil || resp.StatusCode != 200 { continue } if r, parsed := parseRefStatus(body); parsed { // A routing table means members reachable via UDP peers whose detail the // primary endpoint omits — pull the merged roster to enrich them. if len(r.Routing) > 0 { if d := fetchDetailRoster(client, u); d != nil { r.Detail = d } } return r, u, true } } return refStatus{Nodes: map[string]json.RawMessage{}}, "", false } type crawlSeed struct{ tag, host, url string } // crawlNetwork discovers reflectors from the uplink topology, then reads each // reflector's own proxy_status.php for its nodes AND its trunks, following those // trunks' hosts out to further reflectors. Breadth-first by level with bounded // concurrency; cycle-safe (visited hosts), bounded (EXTERNAL_MAX_REFS, depth), // best-effort (unreachable refs listed OK=false, bare-IP trunk hosts skipped). func crawlNetwork(client *http.Client) (sources []extSource, trunks map[string]json.RawMessage, routing map[string]string, generated string, err error) { topoURLs := externalPullURLs() trunks = map[string]json.RawMessage{} routing = map[string]string{} visited := map[string]struct{}{} // Enabled satellite trunks, satellite_id -> friendly callsign (KT4ROY-SAT1), // read from this box's chan_svx_sat config. Used to show the sat trunk under // its real name instead of the raw reflector id (sat403872). satNames := satNamesFromConf(svxSatConf) hostsByKey := map[string]map[string]struct{}{} // IP key -> hostnames seen for it var frontier []crawlSeed lastErr := "" topoOK := false recordHost := func(h string) { h = normalizeHost(h) if h == "" || isIPStr(h) { return } k := hostKey(h) if k == "" { return } if hostsByKey[k] == nil { hostsByKey[k] = map[string]struct{}{} } hostsByKey[k][h] = struct{}{} } addTrunk := func(name string, raw json.RawMessage) { if name == "" { return } if _, dup := trunks[name]; !dup { trunks[name] = raw } } enqueue := func(next *[]crawlSeed, s crawlSeed) { recordHost(s.host) recordHost(hostFromURL(s.url)) key := hostKey(s.host) if key == "" { key = hostKey(hostFromURL(s.url)) } if key == "" { return } if _, seen := visited[key]; seen { return } *next = append(*next, s) } // replayCached folds a reflector's retained last-good contribution into this // cycle's result set (routing, raw trunks, source) and re-seeds its trunk hosts // so the subtree it fronts is still followed. Used when its probe timed out. replayCached := func(e refCacheEntry, next *[]crawlSeed) { for _, en := range e.routing { if _, ok := routing[en.Trunk]; !ok { routing[en.Trunk] = en.Callsign } } for name, raw := range e.trunks { addTrunk(name, raw) } for _, t := range e.src.Trunks { if t.Host != "" { enqueue(next, crawlSeed{tag: t.Host, host: t.Host}) } } sources = append(sources, e.src) } // Seed from the local reflector's OWN live status: every link it reports — // TCP trunks and UDP trunk peers alike — carries the peer host in runtime // data. No trunk mechanism is derived from config-format knowledge here; the // reflector itself is the source of truth for what it is linked to. if body, _, ferr := fetchReflectorStatus(reflectorURL); ferr == nil { if rs, ok := parseRefStatus(body); ok { for _, raw := range rs.Trunks { if h := trunkHost(raw); h != "" { enqueue(&frontier, crawlSeed{tag: h, host: h}) } } for _, raw := range rs.TrunksUDP { if _, _, h := normalizeUDPTrunk(raw); h != "" { enqueue(&frontier, crawlSeed{tag: h, host: h}) } } } } // Seed the crawl with the configured federation reflectors (the Swedish core // etc). They are not in our trunk topology, so without this the crawl never // reaches them; once seeded, the recursive crawl reads each one's /status // (its nodes) and follows its trunks like any other reflector. for _, su := range strings.Split(federationSeeds, ",") { if su = strings.TrimSpace(su); su != "" { enqueue(&frontier, crawlSeed{tag: su, host: hostFromURL(su), url: su}) } } if len(topoURLs) == 0 && len(frontier) == 0 { return nil, nil, nil, "", fmt.Errorf("no uplink derived from %s or %s", svxreflectorConf, reflectorURL) } // Seed from the topology aggregate: the uplink host itself, its sources // (owner-published url), and its trunks (recorded + branched). for _, tu := range topoURLs { p, e := fetchExternalPull(client, tu) if e != nil { lastErr = e.Error() continue } topoOK = true if p.Generated != "" { generated = p.Generated } if uh := hostFromURL(tu); uh != "" { enqueue(&frontier, crawlSeed{tag: uh, host: uh}) } for _, s := range p.Sources { enqueue(&frontier, crawlSeed{tag: s.Tag, host: hostFromURL(s.URL), url: s.URL}) } var tr map[string]json.RawMessage if len(p.Trunks) > 0 && json.Unmarshal(p.Trunks, &tr) == nil { for name, raw := range tr { addTrunk(name, raw) if h := trunkHost(raw); h != "" { enqueue(&frontier, crawlSeed{tag: h, host: h}) } } } } if !topoOK && len(frontier) == 0 { return nil, nil, nil, "", fmt.Errorf("topology unreachable: %s", lastErr) } // Ports to try per host: seeded from the per-server config array (empty by // default — nothing baked in), plus our own reflector's port, then grown with // every real http_port a reached svxref reports (data-driven, no assumption). learnedPorts := map[string]struct{}{} for _, p := range strings.Split(externalStatusPorts, ",") { if p = strings.TrimSpace(p); p != "" { learnedPorts[p] = struct{}{} } } // The port our OWN reflector serves /status on (from REFLECTOR_STATUS_URL) // is a natural candidate for peers running the same software. if u, uerr := url.Parse(reflectorURL); uerr == nil { if p := u.Port(); p != "" { learnedPorts[p] = struct{}{} } } const maxDepth = 4 for depth := 0; depth < maxDepth && len(frontier) > 0; depth++ { ports := make([]string, 0, len(learnedPorts)) for p := range learnedPorts { ports = append(ports, p) } sort.Strings(ports) // Claim this level's hosts (dedup vs visited, respect the ref cap). var batch []crawlSeed for _, s := range frontier { key := hostKey(s.host) if key == "" { key = hostKey(hostFromURL(s.url)) } if key == "" { continue } if _, seen := visited[key]; seen { continue } if externalMaxRefs > 0 && len(visited) >= externalMaxRefs { log.Printf("external crawl: hit EXTERNAL_MAX_REFS=%d — some refs not read", externalMaxRefs) break } visited[key] = struct{}{} batch = append(batch, s) } if len(batch) == 0 { break } // Read the whole level concurrently (bounded). type readResult struct { seed crawlSeed src extSource rs refStatus } results := make([]readResult, len(batch)) sem := make(chan struct{}, 10) var wg sync.WaitGroup for i, s := range batch { wg.Add(1) go func(i int, s crawlSeed) { defer wg.Done() sem <- struct{}{} defer func() { <-sem }() rs, usedURL, ok := readRefStatus(client, s.host, s.url, ports) results[i] = readResult{ seed: s, src: extSource{Tag: s.tag, URL: usedURL, OK: ok, Version: rs.Version, Mode: rs.Mode, Nodes: rs.Nodes, Routing: rs.Routing, Detail: rs.Detail}, rs: rs, } }(i, s) } wg.Wait() // Fold results single-threaded: record trunks (global + per owning server), // then branch to the next level. var next []crawlSeed for _, r := range results { // Learn this reflector's real HTTP port for the next levels. if r.rs.HTTPPort != "" { learnedPorts[r.rs.HTTPPort] = struct{}{} } key := hostKey(r.seed.host) if key == "" { key = hostKey(hostFromURL(r.seed.url)) } // Probe failed this cycle: rather than dropping this reflector (and all // its trunks) — the flicker cause — fall back to its retained last-good // contribution if still fresh, and keep branching from it. if !r.src.OK { if e, ok := freshRefCache(key); ok { replayCached(e, &next) continue } // no fresh cache — record the empty source as before } // Accumulate the routing table (trunk IP -> callsign) for self/identity tagging. for _, e := range r.rs.Routing { if _, ok := routing[e.Trunk]; !ok { routing[e.Trunk] = e.Callsign } } src := r.src src.HTTPPort = r.rs.HTTPPort src.IP = hostKey(r.seed.host) if src.IP == "" { src.IP = hostKey(hostFromURL(r.seed.url)) } if !isIPStr(src.IP) { src.IP = "" // unresolved host — no IP to show } owned := []extTrunkRef{} // never nil → never serialized as JSON null contrib := map[string]json.RawMessage{} // this ref's own raw blobs, kept for retention for name, raw := range r.rs.Trunks { addTrunk(name, raw) contrib[name] = raw h := trunkHost(raw) owned = append(owned, extTrunkRef{Name: name, Host: h, Connected: trunkConnected(raw), Kind: "trunk"}) if h != "" { enqueue(&next, crawlSeed{tag: h, host: h}) } } for _, raw := range r.rs.TrunksUDP { name, norm, h := normalizeUDPTrunk(raw) if name == "" { continue } addTrunk(name, norm) contrib[name] = norm owned = append(owned, extTrunkRef{Name: name, Host: h, Connected: trunkConnected(norm), Kind: "udp"}) if h != "" { enqueue(&next, crawlSeed{tag: h, host: h}) } } for id, sraw := range r.rs.Satellites { var meta struct { Authenticated bool `json:"authenticated"` } _ = json.Unmarshal(sraw, &meta) // Show the ENABLED sat trunk under its real callsign (KT4ROY-SAT1) // from this box's chan_svx_sat config, not the raw reflector id // (sat403872). Not in the config → keep the raw id as a fallback. name := satNames[id] if name == "" { name = id } addTrunk(name, sraw) // so the flat list captures the satellite's nodes[] contrib[name] = sraw owned = append(owned, extTrunkRef{Name: name, Connected: meta.Authenticated, Kind: "satellite"}) } src.Trunks = owned sources = append(sources, src) // Retain this good read so a future timed-out cycle can resurrect it. if r.src.OK && key != "" && externalStaleTTL > 0 { putRefCache(key, refCacheEntry{src: src, trunks: contrib, routing: r.rs.Routing}) } } frontier = next } // Resurrection pass: reflectors reachable recently but not crawled AT ALL this // cycle (their only path in went through a reflector that timed out, so they // were never enqueued) keep their retained last-good trunks until the staleness // window expires. Reflectors already handled above are in `visited` → skipped, // so nothing is duplicated. if externalStaleTTL > 0 { refCacheMu.Lock() var revive []refCacheEntry for k, e := range refCache { if time.Since(e.at) > externalStaleTTL { delete(refCache, k) // prune so the cache stays bounded continue } if _, seen := visited[k]; seen { continue } revive = append(revive, e) } refCacheMu.Unlock() var discard []crawlSeed // last-known picture — do not branch further for _, e := range revive { replayCached(e, &discard) } } // Attach alias hostnames: every other hostname that resolves to this ref's IP // (e.g. reflector-network.se also answers as sm2ampr.net). Done after the crawl // so all hostnames have been recorded. for i := range sources { key := sources[i].IP if key == "" { key = hostKey(sources[i].Tag) } primary := normalizeHost(sources[i].Tag) for h := range hostsByKey[key] { if h != primary { sources[i].Aliases = append(sources[i].Aliases, h) } } sort.Strings(sources[i].Aliases) } // Keep ALL discovered reflectors (online + firewalled/offline). The offline // ones must stay so Topology can show them; Monitor Nodes filters to online // in the frontend. sort.Slice(sources, func(i, j int) bool { return sources[i].Tag < sources[j].Tag }) return sources, trunks, routing, generated, nil } // bridgeHTTP fetches other reflectors' /api/bridge-nodes during the crawl. Skips // TLS verification: this is a best-effort read of a public bridge directory // across many independently-operated servers (varied/self-signed certs). var bridgeHTTP = &http.Client{ Timeout: 2500 * time.Millisecond, Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}, } // crawlBridgeNodes fetches each crawled reflector's OWN /api/bridge-nodes and // unions them into a network-wide bridge directory. A host that doesn't run // svxportal (no endpoint / 404 / timeout) is simply skipped. Every host comes // from the topology the crawl already produced — nothing is hard-coded. func crawlBridgeNodes(sources []extSource) map[string]bridgeInfo { out := map[string]bridgeInfo{} seen := map[string]bool{} var wg sync.WaitGroup var mu sync.Mutex for _, s := range sources { host := hostFromURL(s.URL) // status URL is often empty here... if host == "" { host = s.IP // ...so fall back to the reflector IP (topology-derived) } if host == "" || seen[host] { continue } seen[host] = true hp := host if strings.Contains(host, ":") { hp = "[" + host + "]" // bracket IPv6 literals } wg.Add(1) go func(hp string) { // parallel: whole crawl bounded by one client timeout defer wg.Done() resp, err := bridgeHTTP.Get("https://" + hp + "/svxportal/api/bridge-nodes") if err != nil { return } body, _ := io.ReadAll(io.LimitReader(resp.Body, 256<<10)) resp.Body.Close() if resp.StatusCode != http.StatusOK { return } var m map[string]bridgeInfo if json.Unmarshal(body, &m) != nil { return } mu.Lock() for via, bi := range m { if bi.Node != "" { out[strings.ToUpper(via)] = bi } } mu.Unlock() }(hp) } wg.Wait() return out } // externalPuller periodically crawls the network (topology → per-ref /status → // follow trunks) and stores the result in externalState. Runs as its own // goroutine, off the status-serving hot path. Re-derives the uplink from the // conf each cycle so a peer change is followed without a restart. func externalPuller() { defer func() { if r := recover(); r != nil { log.Printf("externalPuller panic: %v", r) } }() client := &http.Client{Timeout: 2500 * time.Millisecond} // per request; crawl is parallel + best-effort, off the hot path. Higher = fewer transient timeouts (last-good retention covers the rest) loggedNoUplink := false pullOnce := func() { if len(externalPullURLs()) == 0 { if !loggedNoUplink { log.Printf("external crawl: no uplink host in %s and no EXTERNAL_PULL_URL set; disabled", svxreflectorConf) loggedNoUplink = true } return } loggedNoUplink = false sources, trunks, routing, generated, err := crawlNetwork(client) if err != nil { setExternalErr(err.Error()) return } trunksRaw, _ := json.Marshal(trunks) totalNodes := 0 for _, s := range sources { totalNodes += len(s.Nodes) } // Piggyback on the crawl: fetch each discovered reflector's OWN bridge // directory and union it. Hosts come from the topology → nothing is // hard-coded. Best-effort; keep last-good on an empty round. bridges := crawlBridgeNodes(sources) externalState.mu.Lock() externalState.sources = sources externalState.trunks = trunksRaw externalState.routing = routing externalState.generated = generated externalState.lastOKAt = time.Now() externalState.lastErr = "" if len(bridges) > 0 { externalState.bridgeNodes = bridges } externalState.mu.Unlock() if debug { log.Printf("DEBUG external crawl: refs=%d nodes=%d trunks=%d", len(sources), totalNodes, len(trunks)) } } pullOnce() t := time.NewTicker(externalPullInterval) defer t.Stop() for range t.C { pullOnce() } } func setExternalErr(msg string) { externalState.mu.Lock() externalState.lastErr = msg externalState.mu.Unlock() if debug { log.Printf("DEBUG external pull failed: %s (url=%s)", msg, externalPullURL) } } // tglistPuller keeps the canonical TG->system map fresh from the shared tglist // (svx-web /tglist.php). Mirrors externalPuller: fetch on a gentle ticker, keep // last-good on failure, cache to disk so names survive a restart before the // first successful fetch. URL empty = feature off — the per-box nodes[].system // stays the only source, so nothing changes until a box opts in via config. func tglistPuller(ctx context.Context, c *config.Config) { defer func() { if r := recover(); r != nil { log.Printf("tglistPuller panic: %v", r) } }() file := strings.TrimSpace(c.Tglist.File) url := strings.TrimSpace(c.Tglist.URL) if file == "" && url == "" { log.Printf("tglist: no file/url set; Location uses per-box nodes[].system only") return } cacheFile := strings.TrimSpace(c.Tglist.CacheFile) if cacheFile == "" { cacheFile = "/var/lib/svxdr/tglist.json" } interval := c.Tglist.RefreshInterval.D() if interval <= 0 { interval = 15 * time.Minute } client := &http.Client{Timeout: 4 * time.Second} // Parse a /tglist.php body into the TG->system map (system, else name). apply := func(body []byte) int { var doc struct { TGs []struct { TG int `json:"tg"` Name string `json:"name"` System string `json:"system"` } `json:"tgs"` } if err := json.Unmarshal(body, &doc); err != nil { return -1 } m := make(map[int]string, len(doc.TGs)) for _, e := range doc.TGs { if e.TG <= 0 { continue } s := strings.TrimSpace(e.System) if s == "" { s = strings.TrimSpace(e.Name) } if s != "" { m[e.TG] = s } } setTglistSystem(m) return len(m) } // Cold start: seed from the disk cache so names are present even before the // first fetch (and if the list host is unreachable at boot). if b, err := os.ReadFile(cacheFile); err == nil && len(b) > 0 { if n := apply(b); n >= 0 { log.Printf("tglist: seeded %d TG names from cache %s", n, cacheFile) } } pullOnce := func() { // SVX source of truth is a local JSON file — read it directly, no HTTP. if file != "" { body, err := os.ReadFile(file) if err != nil { if debug { log.Printf("DEBUG tglist read failed: %v (file=%s)", err, file) } return } if n := apply(body); n >= 0 && debug { log.Printf("DEBUG tglist: %d TG names from %s", n, file) } return } req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { return } resp, err := client.Do(req) if err != nil { if debug { log.Printf("DEBUG tglist fetch failed: %v (url=%s)", err, url) } return } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { if debug { log.Printf("DEBUG tglist fetch status %d (url=%s)", resp.StatusCode, url) } return } body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) if err != nil { return } n := apply(body) if n < 0 { return // parse error: keep last-good } if err := os.MkdirAll(filepath.Dir(cacheFile), 0o755); err == nil { _ = os.WriteFile(cacheFile, body, 0o644) } if debug { log.Printf("DEBUG tglist: %d TG names from %s", n, url) } } pullOnce() t := time.NewTicker(interval) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: pullOnce() } } } // ---- Complete network structure (nodelist collects everything; the website // picks what it needs). Every slice is always allocated so nothing serializes as // JSON null. Nothing is stripped: online + offline reflectors are all kept, tagged. type netNode struct { Callsign string `json:"callsign"` TG int `json:"tg"` RestrictedTG bool `json:"restrictedTG"` Location string `json:"location"` NodeClass string `json:"nodeClass"` SW string `json:"sw"` Sysop string `json:"sysop"` MonitoredTGs []int `json:"monitoredTGs"` IsTalker bool `json:"isTalker"` RxFreq string `json:"rxFreq"` TxFreq string `json:"txFreq"` RxCtcss string `json:"rxCtcss"` // node's REPORT_CTCSS (the subtone it transmits) TxCtcss string `json:"txCtcss"` // access tone selecting the current TG } type netTrunkRef struct { Name string `json:"name"` Host string `json:"host"` Connected bool `json:"connected"` Kind string `json:"kind"` } type netReflector struct { Type string `json:"type"` // "local" | "reflector" Self bool `json:"self"` Uplink bool `json:"uplink"` Name string `json:"name"` Hostname string `json:"hostname"` IP string `json:"ip"` Port string `json:"port"` Online bool `json:"online"` // up in the network (HTTP-read OR a trunk to it is connected) Firewalled bool `json:"firewalled"` // online via trunk but HTTP status unreachable (no node detail) Version string `json:"version"` Mode string `json:"mode"` Aliases []string `json:"aliases"` Nodes []netNode `json:"nodes"` Trunks []netTrunkRef `json:"trunks"` } type netTrunk struct { Name string `json:"name"` Owner string `json:"owner"` Host string `json:"host"` Connected bool `json:"connected"` NodeCount int `json:"node_count"` Kind string `json:"kind"` Nodes []netNode `json:"nodes"` // callsigns this trunk carries (the "external" node list) } type networkDoc struct { Generated string `json:"generated"` Reflectors []netReflector `json:"reflectors"` Trunks []netTrunk `json:"trunks"` } var ( networkMu sync.RWMutex networkJSON = []byte(`{"generated":"","reflectors":[],"trunks":[]}`) lastNetworkHash string networkJSONPath = env("NETWORK_JSON_PATH", "/var/www/html/svxportal/network.json") ) // writeMaybeGzip writes b with a 200, gzip-compressing when the client accepts it. // The status/network blobs are large (~900 KB) and served to every browser on each // update, so gzip (~6x smaller) is a major bandwidth win. The caller sets Content-Type. func writeMaybeGzip(w http.ResponseWriter, r *http.Request, b []byte) { if strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") { w.Header().Set("Content-Encoding", "gzip") w.Header().Set("Vary", "Accept-Encoding") w.WriteHeader(200) gz := gzip.NewWriter(w) _, _ = gz.Write(b) _ = gz.Close() return } w.WriteHeader(200) _, _ = w.Write(b) } // rawScalarStr returns a JSON scalar (string or number) as a trimmed string. func rawScalarStr(r json.RawMessage) string { if len(r) == 0 { return "" } var s string if json.Unmarshal(r, &s) == nil { return strings.TrimSpace(s) } return strings.TrimSpace(strings.Trim(string(r), `"`)) } // toNetNode extracts the fields the website needs from a raw node object. func toNetNode(callsign string, raw json.RawMessage) netNode { var n struct { Callsign string `json:"callsign"` TG int `json:"tg"` RestrictedTG bool `json:"restrictedTG"` NodeLocation string `json:"nodeLocation"` NodeClass string `json:"nodeClass"` SW string `json:"sw"` SwVer string `json:"swVer"` Sysop string `json:"sysop"` MonitoredTGs []int `json:"monitoredTGs"` IsTalker bool `json:"isTalker"` RXFREQ json.RawMessage `json:"RXFREQ"` TXFREQ json.RawMessage `json:"TXFREQ"` ReportCtcss json.RawMessage `json:"reportCtcss"` ToneToTG map[string]int `json:"toneToTalkgroup"` QTH []struct { Name string `json:"name"` RX map[string]struct { Freq json.RawMessage `json:"freq"` } `json:"rx"` TX map[string]struct { Freq json.RawMessage `json:"freq"` } `json:"tx"` } `json:"qth"` } _ = json.Unmarshal(raw, &n) cs := callsign if cs == "" { cs = n.Callsign } loc := n.NodeLocation if loc == "" { for _, q := range n.QTH { if q.Name != "" { loc = q.Name break } } } // Frequencies: top-level RXFREQ/TXFREQ, else the first qth rx/tx freq. rxFreq, txFreq := rawScalarStr(n.RXFREQ), rawScalarStr(n.TXFREQ) if rxFreq == "" || txFreq == "" { for _, q := range n.QTH { if rxFreq == "" { for _, p := range q.RX { if f := rawScalarStr(p.Freq); f != "" { rxFreq = f break } } } if txFreq == "" { for _, p := range q.TX { if f := rawScalarStr(p.Freq); f != "" { txFreq = f break } } } if rxFreq != "" && txFreq != "" { break } } } // Subtones: RX = the node's REPORT_CTCSS; TX = the access tone selecting the // current TG (from toneToTalkgroup). rxCtcss := rawScalarStr(n.ReportCtcss) txCtcss := "" if n.TG != 0 { for tone, tg := range n.ToneToTG { if tg == n.TG { txCtcss = tone break } } } sw := strings.TrimSpace(n.SW + " " + n.SwVer) mt := n.MonitoredTGs if mt == nil { mt = []int{} } return netNode{Callsign: cs, TG: n.TG, RestrictedTG: n.RestrictedTG, Location: loc, NodeClass: n.NodeClass, SW: sw, Sysop: n.Sysop, MonitoredTGs: mt, IsTalker: n.IsTalker, RxFreq: rxFreq, TxFreq: txFreq, RxCtcss: rxCtcss, TxCtcss: txCtcss} } // aliasesNorm returns the normalized-host form of each alias. func aliasesNorm(aliases []string) []string { out := make([]string, 0, len(aliases)) for _, a := range aliases { out = append(out, normalizeHost(a)) } return out } // localReflectorMeta reads the local reflector's name (SERVER_CERT ORG) and the // identifiers (hostname/IP) that mark a crawled ref as "self". func localReflectorMeta() (name string, selfKeys map[string]struct{}) { selfKeys = map[string]struct{}{} name = "LOCAL" f, err := os.Open(svxreflectorConf) if err != nil { return } defer f.Close() sc := bufio.NewScanner(f) inCert := false for sc.Scan() { line := strings.TrimSpace(sc.Text()) if strings.HasPrefix(line, "[") && strings.HasSuffix(line, "]") { inCert = strings.EqualFold(line[1:len(line)-1], "SERVER_CERT") continue } if !inCert { continue } i := strings.IndexByte(line, '=') if i < 0 { continue } k := strings.ToUpper(strings.TrimSpace(line[:i])) v := strings.TrimSpace(line[i+1:]) switch k { case "ORG": if v != "" { name = v } case "COMMON_NAME": if v != "" { selfKeys[normalizeHost(v)] = struct{}{} if ip := hostKey(v); ip != "" { selfKeys[ip] = struct{}{} } } case "SUBJECT_ALT_NAME": for _, part := range strings.Split(v, ",") { part = strings.TrimSpace(part) up := strings.ToUpper(part) if strings.HasPrefix(up, "DNS:") { h := normalizeHost(part[4:]) selfKeys[h] = struct{}{} if ip := hostKey(h); ip != "" { selfKeys[ip] = struct{}{} } } else if strings.HasPrefix(up, "IP:") { selfKeys[strings.TrimSpace(part[3:])] = struct{}{} } } } } return } // buildNetworkDoc assembles the complete structure from the local reflector // status (body) + the crawl (externalState). Strips nothing; tags everything. func buildNetworkDoc(body []byte) networkDoc { externalState.mu.RLock() sources := externalState.sources rawTrunks := externalState.trunks routing := externalState.routing generated := externalState.generated externalState.mu.RUnlock() doc := networkDoc{Generated: generated, Reflectors: []netReflector{}, Trunks: []netTrunk{}} localName, selfKeys := localReflectorMeta() uplinks := map[string]struct{}{} for _, h := range uplinkHostsFromConf(svxreflectorConf) { uplinks[normalizeHost(h)] = struct{}{} } // Local reflector entry (SA7AUX): nodes without remoteTrunk + local trunks. var top map[string]json.RawMessage _ = json.Unmarshal(body, &top) localNodes := []netNode{} if raw, ok := top["nodes"]; ok { var m map[string]json.RawMessage if json.Unmarshal(raw, &m) == nil { for cs, nr := range m { var chk struct { RemoteTrunk json.RawMessage `json:"remoteTrunk"` } _ = json.Unmarshal(nr, &chk) if len(chk.RemoteTrunk) > 0 { continue } localNodes = append(localNodes, toNetNode(cs, nr)) } } } sort.Slice(localNodes, func(i, j int) bool { return localNodes[i].Callsign < localNodes[j].Callsign }) localTrunks := []netTrunkRef{} if raw, ok := top["trunks"]; ok { var m map[string]json.RawMessage if json.Unmarshal(raw, &m) == nil { for name, tr := range m { localTrunks = append(localTrunks, netTrunkRef{Name: name, Host: trunkHost(tr), Connected: trunkConnected(tr), Kind: "trunk"}) } } } if raw, ok := top["trunks_UDP"]; ok { var arr []json.RawMessage if json.Unmarshal(raw, &arr) == nil { for _, tr := range arr { name, norm, h := normalizeUDPTrunk(tr) if name == "" { continue } localTrunks = append(localTrunks, netTrunkRef{Name: name, Host: h, Connected: trunkConnected(norm), Kind: "udp"}) } } } if raw, ok := top["satellites"]; ok { var m map[string]struct { Authenticated bool `json:"authenticated"` } if json.Unmarshal(raw, &m) == nil { satNames := satNamesFromConf(svxSatConf) for id, sv := range m { // Real callsign (KT4ROY-SAT1) from chan_svx_sat, not the raw id. name := satNames[id] if name == "" { name = id } localTrunks = append(localTrunks, netTrunkRef{Name: name, Connected: sv.Authenticated, Kind: "satellite"}) } } } doc.Reflectors = append(doc.Reflectors, netReflector{ Type: "local", Self: true, Name: localName, Online: true, Aliases: []string{}, Nodes: localNodes, Trunks: localTrunks, }) // Parse the merged trunk map once: for node counts (flat list below) and to // know which hosts have a connected trunk (up in the network even if their // HTTP status is firewalled from here). var trmap map[string]json.RawMessage _ = json.Unmarshal(rawTrunks, &trmap) if trmap == nil { trmap = map[string]json.RawMessage{} } // The local reflector's own /status carries each trunk's peer nodes (pushed via // the reflector's trunk protocol) — take them straight from there. We poll NO // ONE for network info; the local status is the single source. if bt, ok := top["trunks"]; ok { var bm map[string]json.RawMessage if json.Unmarshal(bt, &bm) == nil { for name, raw := range bm { if _, dup := trmap[name]; !dup { trmap[name] = raw } } } } // Satellite trunks: their nodes live in the body's "satellites" block, keyed by // sat id; index them under the same name localTrunks uses so their nodes show. if bs, ok := top["satellites"]; ok { var sm map[string]json.RawMessage if json.Unmarshal(bs, &sm) == nil { satNames := satNamesFromConf(svxSatConf) for id, raw := range sm { name := satNames[id] if name == "" { name = id } if _, dup := trmap[name]; !dup { trmap[name] = raw } } } } connectedHosts := map[string]struct{}{} for _, tr := range trmap { var m struct { Host string `json:"host"` Connected bool `json:"connected"` } if json.Unmarshal(tr, &m) == nil && m.Connected && m.Host != "" { connectedHosts[normalizeHost(m.Host)] = struct{}{} } } // Route-based node attribution (mirrors SvxReflectorDashboard's // merge_trunk_status_nodes): every crawled reflector's Routing_table maps a // callsign to the trunk IP it is reachable via. Group by IP so trunks and // reflectors whose own node list is unreadable (firewalled UDP peers) still // show their members. Deduped per IP; our own callsign is never added (no // self-dupe). // Full per-node detail (location/freq/sw/sysop) pulled from reflectors' merged // rosters — used to enrich route nodes the Routing_table only names sparsely. detailByCallsign := map[string]json.RawMessage{} for _, s := range sources { for cs, nr := range s.Detail { if _, ok := detailByCallsign[cs]; !ok { detailByCallsign[cs] = nr } } } routeByIP := map[string][]netNode{} routeSeen := map[string]map[string]struct{}{} for _, s := range sources { for _, e := range s.Routing { ip := normalizeHost(e.Trunk) if ip == "" || e.Callsign == "" { continue } if localName != "" && strings.EqualFold(e.Callsign, localName) { continue } if routeSeen[ip] == nil { routeSeen[ip] = map[string]struct{}{} } if _, dup := routeSeen[ip][e.Callsign]; dup { continue } routeSeen[ip][e.Callsign] = struct{}{} // Prefer the full-detail node from the merged roster; fall back to the // sparse Routing_table fields (callsign + tg + monitored TGs). if draw, ok := detailByCallsign[e.Callsign]; ok { routeByIP[ip] = append(routeByIP[ip], toNetNode(e.Callsign, draw)) continue } mon := e.MonitoredTGs if mon == nil { mon = []int{} } routeByIP[ip] = append(routeByIP[ip], netNode{Callsign: e.Callsign, TG: e.TG, MonitoredTGs: mon}) } } // Crawled reflectors — all of them, tagged (nothing dropped). for _, s := range sources { nodes := []netNode{} for cs, nr := range s.Nodes { nodes = append(nodes, toNetNode(cs, nr)) } // Fill members we can't read directly (firewalled peers) from the routing // tables of reflectors that DO see this one — matched by IP / tag. Deduped // by callsign so a directly-read node is never duplicated. seenN := map[string]struct{}{} for _, n := range nodes { seenN[n.Callsign] = struct{}{} } for _, key := range []string{normalizeHost(s.IP), normalizeHost(s.Tag)} { if key == "" { continue } for _, rn := range routeByIP[key] { if _, ok := seenN[rn.Callsign]; ok { continue } seenN[rn.Callsign] = struct{}{} nodes = append(nodes, rn) } } sort.Slice(nodes, func(i, j int) bool { return nodes[i].Callsign < nodes[j].Callsign }) trunks := []netTrunkRef{} for _, t := range s.Trunks { trunks = append(trunks, netTrunkRef{Name: t.Name, Host: t.Host, Connected: t.Connected, Kind: t.Kind}) } aliases := s.Aliases if aliases == nil { aliases = []string{} } _, isSelf := selfKeys[s.IP] if !isSelf { _, isSelf = selfKeys[normalizeHost(s.Tag)] } // Routing table: this ref's trunk IP maps to our own callsign → it's us. if !isSelf && localName != "" { if strings.EqualFold(routing[s.IP], localName) || strings.EqualFold(routing[normalizeHost(s.Tag)], localName) { isSelf = true } } _, isUplink := uplinks[normalizeHost(s.Tag)] if !isUplink { _, isUplink = uplinks[normalizeHost(hostFromURL(s.URL))] } hostname := hostFromURL(s.URL) if hostname == "" && !isIPStr(s.Tag) { hostname = s.Tag } // Online = HTTP-read OR a trunk to this reflector is connected. If it's up // only via a connected trunk (HTTP unreachable), it's firewalled. online := s.OK firewalled := false if !online { for _, k := range append([]string{normalizeHost(s.Tag), normalizeHost(hostname), normalizeHost(s.IP)}, aliasesNorm(aliases)...) { if k == "" { continue } if _, ok := connectedHosts[k]; ok { online = true firewalled = true break } } } doc.Reflectors = append(doc.Reflectors, netReflector{ Type: "reflector", Self: isSelf, Uplink: isUplink, Name: s.Tag, Hostname: hostname, IP: s.IP, Port: s.HTTPPort, Online: online, Firewalled: firewalled, Version: s.Version, Mode: s.Mode, Aliases: aliases, Nodes: nodes, Trunks: trunks, }) } // Flat list of ALL links in the network — trunks, UDP peers AND satellites — // built from the reflectors' own link lists (deduped by name), with owner + // node count. This includes satellites, which the merged trunk map omits. seenT := map[string]struct{}{} for _, r := range doc.Reflectors { for _, t := range r.Trunks { if _, ok := seenT[t.Name]; ok { continue } seenT[t.Name] = struct{}{} tnodes := []netNode{} if raw, ok := trmap[t.Name]; ok { var m struct { Nodes []json.RawMessage `json:"nodes"` } if json.Unmarshal(raw, &m) == nil { for _, nr := range m.Nodes { tnodes = append(tnodes, toNetNode("", nr)) } } } // A UDP peer exposes no nodes[] inline; fill its members from the owning // reflector's Routing_table (callsigns reachable via its host IP). Deduped. seenTN := map[string]struct{}{} for _, n := range tnodes { seenTN[n.Callsign] = struct{}{} } for _, rn := range routeByIP[normalizeHost(t.Host)] { if _, ok := seenTN[rn.Callsign]; ok { continue } seenTN[rn.Callsign] = struct{}{} tnodes = append(tnodes, rn) } sort.Slice(tnodes, func(i, j int) bool { return tnodes[i].Callsign < tnodes[j].Callsign }) doc.Trunks = append(doc.Trunks, netTrunk{ Name: t.Name, Owner: r.Name, Host: t.Host, Connected: t.Connected, NodeCount: len(tnodes), Kind: t.Kind, Nodes: tnodes, }) } } sort.Slice(doc.Trunks, func(i, j int) bool { return doc.Trunks[i].Name < doc.Trunks[j].Name }) return doc } // writeNetworkFile atomically writes b to networkJSONPath (temp + rename) when it // changed, so external fetchers never see a half-written file. func writeNetworkFile(b []byte) { if networkJSONPath == "" { return } dir := filepath.Dir(networkJSONPath) tmp, err := os.CreateTemp(dir, ".network-*.json") if err != nil { return } tmpName := tmp.Name() if _, err := tmp.Write(b); err != nil { tmp.Close() os.Remove(tmpName) return } tmp.Close() if err := os.Rename(tmpName, networkJSONPath); err != nil { os.Remove(tmpName) } } // injectExternal adds ext_sources and network_trunks (from externalState) as // top-level fields to the served status JSON. If no external data is available // yet, body is returned unchanged so local nodes/trunks still go out. func injectExternal(body []byte) []byte { // Build the complete network structure and publish it (served at /api/network // and written to networkJSONPath for other systems to fetch). doc := buildNetworkDoc(body) nb, nerr := json.Marshal(doc) if nerr == nil { networkMu.Lock() networkJSON = nb networkMu.Unlock() if h := hash(nb); h != lastNetworkHash { lastNetworkHash = h writeNetworkFile(nb) } } var top map[string]json.RawMessage if json.Unmarshal(body, &top) != nil { return body } // The complete structure the frontend reads. The old flat ext_sources / // network_trunks fields (unused by the frontend) are intentionally NOT emitted // — they were ~66% of the served blob. Other systems fetch /api/network. if nerr == nil { top["network"] = nb } if bb, err := json.Marshal(top); err == nil { return bb } return body } const ( heartbeatKey = "/registry/svxlink/nodelist/heartbeat" ) type statusEnvelope struct { Nodes map[string]json.RawMessage `json:"nodes"` Trunks map[string]trunkStatus `json:"trunks"` Satellites map[string]satelliteStatus `json:"satellites"` } // satelliteStatus mirrors a satellite trunk (chan_svx_sat / chan_pbx_svx). Its // nodes[] carry a nodeClass so a PBX gateway talker can be shown as // " via " rather than the raw gateway node name. type satelliteStatus struct { Nodes []struct { Callsign string `json:"callsign"` TG int `json:"tg"` IsTalker bool `json:"isTalker"` NodeClass string `json:"nodeClass"` NodeLocation string `json:"nodeLocation"` SW string `json:"sw"` } `json:"nodes"` } // trunkStatus mirrors the per-trunk JSON emitted by the reflector // (SatelliteLink/TrunkLink statusJson): active_talkers maps a TG to the real // remote talker callsign, and nodes[] lists partner nodes with isTalker. type trunkStatus struct { ActiveTalkers map[string]string `json:"active_talkers"` Nodes []struct { Callsign string `json:"callsign"` TG int `json:"tg"` IsTalker bool `json:"isTalker"` } `json:"nodes"` } type nodeInfo struct { ToneToTalkgroup map[string]int `json:"toneToTalkgroup"` QTH []struct { Name string `json:"name"` RX map[string]struct { Freq float64 `json:"freq"` Name string `json:"name"` Active bool `json:"active"` Siglev float64 `json:"siglev"` } `json:"rx"` TX map[string]struct { Freq float64 `json:"freq"` } `json:"tx"` } `json:"qth"` RX float64 `json:"rx"` TX float64 `json:"tx"` RxFreq float64 `json:"rxFreq"` TxFreq float64 `json:"txFreq"` IsTalker bool `json:"isTalker"` TG int `json:"tg"` NodeLocation string `json:"nodeLocation"` Sysop string `json:"sysop"` } // live holds the real-callsign lookup maps maintained by the poller and read // by the log tailer: // - tgTalker: TG -> real remote talker callsign (from trunks[].active_talkers) // - nodeSysop: node-name -> operator callsign (from nodes[].sysop) // - tgMeta: TG -> talker meta published on MQTT (talker//meta) by an // external source such as astmeta (AllStar bridge talker id) var live = struct { mu sync.RWMutex tgTalker map[int]string tgTalkerAt map[int]int64 // TG -> unix ts tgTalker[tg] last had a value (sticky window) nodeSysop map[string]string tgMeta map[int]talkerMeta asNodes map[string]allstarNode viaNode map[string]bridgeInfo // UPPER(via callsign) -> bridge target, long-lived tgLoc map[int]string // TG -> gateway nodeLocation (e.g. SA7AUX-PBX), for the Location column tgMode map[int]string // TG -> Backend/mode of the gateway node (e.g. PBX), for the Backend column }{tgTalker: map[int]string{}, tgTalkerAt: map[int]int64{}, nodeSysop: map[string]string{}, tgMeta: map[int]talkerMeta{}, asNodes: map[string]allstarNode{}, viaNode: map[string]bridgeInfo{}, tgLoc: map[int]string{}, tgMode: map[int]string{}} // locForTG returns the gateway nodeLocation currently mapped to a TG (empty if none). func locForTG(tg int) string { // FROZEN DESIGN: the tglist is the SOLE source of the Location shown in // Lastheard. No live gateway location, no astmeta, no nodes[].system — every // talker on a TG shows exactly the tglist name for that TG (blank if the TG // is not in the list), so the display is identical on every box. return systemForTG(tg) } // modeForTG returns the gateway Backend/mode currently mapped to a TG (empty if none). func modeForTG(tg int) string { live.mu.RLock() defer live.mu.RUnlock() return live.tgMode[tg] } // bridgeInfo is what a bridge login (via) speaks through: the AllStar node it is // connected to and that node's system name (e.g. 45192 / "East Coast Reflector"). // Long-lived + redis-persisted so the web can show "connected to " and // the connected-nodes button on the bridge row even when it is idle. type bridgeInfo struct { Node string `json:"node"` System string `json:"system"` } // allstarNode is one station/system heard through the bridge via the // talker-meta ingest. Kept ALLSTAR_NODES_TTL from the last keyup and // published in reflector-status "nodes" format on /api/allstar-nodes. type allstarNode struct { Callsign string Node string System string Via string TG int At time.Time } // talkerMeta is one talker-meta entry (fed by POST /api/talker-meta): who is // really talking on the TG and from which node/system. Generic across // sources: the AllStar bridge fills it via the svxlink TCL hook (node = // AllStar node number), a DMR/NXDN bridge can fill the same fields (node = // DMR id, resolved against the radioid tables in redis). type talkerMeta struct { Callsign string `json:"callsign"` Node string `json:"node"` System string `json:"system"` Via string `json:"via"` // reflector callsign the bridge speaks through Network string `json:"network"` // ingress mode: AllStar / DMR / DMR/YSF … At time.Time } // talkerMetaFor returns the fresh talker meta for a TG, if any. logName is // the talker name as it appears in the reflector log: when the meta carries // a "via" callsign it only applies to that talker, so lingering meta can // never mislabel an unrelated station on the same TG. func talkerMetaFor(tg int, logName string) (talkerMeta, bool) { live.mu.RLock() defer live.mu.RUnlock() m, ok := live.tgMeta[tg] if !ok || m.Callsign == "" || time.Since(m.At) > talkerMetaTTL { return talkerMeta{}, false } // The meta's via identifies the talker's bridge. Prefer the per-TG via // configured in svxdr.json (nodes[].via) so the callsign/via is defined in // ONE place and need NOT equal the reflector-log talker name (svx_sat.conf // callsign). Only when no per-TG via is configured do we fall back to the // old log-name match (backward compat for boxes without a nodes[] config). if anyViaForTG(tg) { if m.Via != "" && !viaConfiguredForTG(tg, m.Via) { return talkerMeta{}, false } } else if m.Via != "" && m.Via != logName { return talkerMeta{}, false } return m, true } // resolveTalker maps a node/trunk name (as it appears in the reflector log) to // the real operator callsign, returning the trunk/node name separately when it // differs so the UI can show both. For local nodes, fresh talker meta // (e.g. the AllStar system behind a USRP bridge) wins over the sysop mapping. func resolveTalker(nodeName string, tg int, isTrunk bool) (callsign, trunk string) { callsign = nodeName live.mu.RLock() if isTrunk { if c := live.tgTalker[tg]; c != "" { callsign = c } } else if m, ok := live.tgMeta[tg]; ok && m.Callsign != "" && time.Since(m.At) <= talkerMetaTTL && (m.Via == "" || m.Via == nodeName) { callsign = m.Callsign } else if s := live.nodeSysop[nodeName]; s != "" { callsign = s } live.mu.RUnlock() if callsign != nodeName { trunk = nodeName } return } func isDigits(s string) bool { if s == "" { return false } for _, r := range s { if r < '0' || r > '9' { return false } } return true } // isLoopbackAddr reports whether a net/http RemoteAddr ("ip:port") is loopback // (127.0.0.0/8 or ::1). Used to let same-box bridges POST talker-meta untokened. func isLoopbackAddr(remote string) bool { host, _, err := net.SplitHostPort(remote) if err != nil { host = remote } ip := net.ParseIP(host) return ip != nil && ip.IsLoopback() } // injectTalkerMeta adds a top-level talker_meta object (tg -> fresh meta) to // the served status JSON so the web can show the real talker behind a bridge // node. The web applies the same via-gating as lastheard: the meta only // labels the node whose callsign matches via. func injectTalkerMeta(body []byte) []byte { type metaOut struct { Callsign string `json:"callsign"` Node string `json:"node"` System string `json:"system"` Via string `json:"via"` } out := map[string]metaOut{} bridge := map[string]bridgeInfo{} // via callsign -> {node, system} (for the // web's per-bridge "connected to" + button) // Network-wide bridges first (crawled from every reflector's own // /api/bridge-nodes), so any box shows a bridge that lives on another // server; the LOCAL map overrides on conflict (it is authoritative). externalState.mu.RLock() for via, bi := range externalState.bridgeNodes { bridge[via] = bi } externalState.mu.RUnlock() live.mu.RLock() for tg, m := range live.tgMeta { if m.Callsign == "" || time.Since(m.At) > talkerMetaTTL { continue } out[strconv.Itoa(tg)] = metaOut{m.Callsign, m.Node, m.System, m.Via} } for via, bi := range live.viaNode { bridge[via] = bi } live.mu.RUnlock() if len(out) == 0 && len(bridge) == 0 { return body } var top map[string]json.RawMessage if json.Unmarshal(body, &top) != nil { return body } if len(out) > 0 { if b, err := json.Marshal(out); err == nil { top["talker_meta"] = b } } if len(bridge) > 0 { if b, err := json.Marshal(bridge); err == nil { top["bridge_nodes"] = b } } bb, err := json.Marshal(top) if err != nil { return body } return bb } type talkerSession struct { StartTs int64 TG int Location string RxName string Siglev float64 } type lastheardEvent struct { Ts int64 `json:"ts"` Callsign string `json:"callsign"` Trunk string `json:"trunk,omitempty"` TG int `json:"tg"` Duration int64 `json:"duration_s"` Location string `json:"location,omitempty"` RxName string `json:"rx_name,omitempty"` Siglev float64 `json:"siglev,omitempty"` Source string `json:"source"` // Source-network identity from talker meta (astmeta for AllStar, the // DMR bridge later): node/ID number and human-readable system name. // Empty for plain svxlink traffic — the UI shows them as blank columns. Node string `json:"node,omitempty"` System string `json:"system,omitempty"` // Ingress mode: HOW the talker enters — "AllStar", "SVXLink", "trunk", // and later "DMR", "DMR/YSF", "DMR/OB"… from the callmgr bridge. The web // shows this as the Mode column (replacing the old System name). Always // set: a row with no meta defaults to "trunk" or "SVXLink" by source. Network string `json:"network,omitempty"` } type runtimeState struct { mu sync.RWMutex StartedAt time.Time LastFetchOKAt time.Time LastFetchURL string LastFetchBytes int LastFetchErr string LastParsedNodes int LastStatusHash string LastRedisOKAt time.Time LastRedisErr string } var st = &runtimeState{ StartedAt: time.Now(), } // ---- Fas 2: federation gap detector ------------------------------------------ // Compares the network-wide aggregate (FEDERATION_AGGREGATE_URL, sa2blv's // reflector-network) against what our OWN recursive crawl + MQTT actually cover. // Purely diagnostic: it names the reflectors/callsigns we are NOT seeing so they // can be added to FEDERATION_SEEDS. Nothing here is merged into the served roster // — our own fresh data stays authoritative. var federationGap = struct { mu sync.RWMutex updatedAt time.Time aggTotal int haveTotal int missingCS int missingRefs map[string]int // reflector -> count of its nodes we do not cover err string }{missingRefs: map[string]int{}} // ourCoveredCallsigns returns the set of callsigns we already see, from the // recursive crawl (every reflector's /status nodes) and every MQTT source. func ourCoveredCallsigns() map[string]struct{} { have := map[string]struct{}{} externalState.mu.RLock() for _, s := range externalState.sources { for cs := range s.Nodes { have[cs] = struct{}{} } } externalState.mu.RUnlock() mqttState.mu.RLock() for _, m := range mqttState.bySrc { for cs := range m { have[cs] = struct{}{} } } mqttState.mu.RUnlock() return have } func setFederationGapErr(msg string) { federationGap.mu.Lock() federationGap.err = msg federationGap.updatedAt = time.Now() federationGap.mu.Unlock() if debug { log.Printf("DEBUG federation gap fetch failed: %s", msg) } } func federationGapOnce(client *http.Client) { resp, err := client.Get(federationAggURL) if err != nil { setFederationGapErr(err.Error()) return } body, _ := io.ReadAll(io.LimitReader(resp.Body, 16<<20)) resp.Body.Close() if resp.StatusCode != http.StatusOK { setFederationGapErr(fmt.Sprintf("HTTP %d", resp.StatusCode)) return } // Aggregate is an object keyed by callsign; we only need each node's reflector. var agg map[string]struct { Reflector string `json:"reflector"` } if json.Unmarshal(body, &agg) != nil { setFederationGapErr("aggregate is not a callsign->node object") return } // Feed the aggregate into the All-Nodes flat layer so Monitor Nodes covers // the whole network (real reflector names), not just our own crawl. am := make(map[string]flatNode, len(agg)) for cs, n := range agg { r := strings.TrimSpace(n.Reflector) if r == "" { continue } am[strings.ToUpper(cs)] = flatNode{Callsign: cs, Ref: r} } setAggNodes(am) have := ourCoveredCallsigns() missingRefs := map[string]int{} missing := 0 for cs, n := range agg { if _, ok := have[cs]; ok { continue } missing++ ref := strings.TrimSpace(n.Reflector) if ref == "" { ref = "(unknown)" } missingRefs[ref]++ } federationGap.mu.Lock() federationGap.updatedAt = time.Now() federationGap.aggTotal = len(agg) federationGap.haveTotal = len(have) federationGap.missingCS = missing federationGap.missingRefs = missingRefs federationGap.err = "" federationGap.mu.Unlock() if missing > 0 { type kv struct { ref string n int } top := make([]kv, 0, len(missingRefs)) for r, n := range missingRefs { top = append(top, kv{r, n}) } sort.Slice(top, func(i, j int) bool { return top[i].n > top[j].n }) parts := make([]string, 0, len(top)) for i, t := range top { if i >= 12 { parts = append(parts, fmt.Sprintf("…+%d more", len(top)-i)) break } parts = append(parts, fmt.Sprintf("%s(%d)", t.ref, t.n)) } log.Printf("federation gap: %d/%d aggregate callsigns not covered — missing reflectors: %s (add to FEDERATION_SEEDS)", missing, len(agg), strings.Join(parts, ", ")) } else if debug { log.Printf("federation gap: 0 — we cover all %d aggregate callsigns", len(agg)) } } func federationGapLoop() { if federationAggURL == "" { log.Printf("federation gap detector disabled (FEDERATION_AGGREGATE_URL empty)") return } client := &http.Client{Timeout: 20 * time.Second} time.Sleep(30 * time.Second) // let the crawl populate before the first diff federationGapOnce(client) t := time.NewTicker(federationAggInterval) defer t.Stop() for range t.C { federationGapOnce(client) } } // Run is the nodelist subsystem entrypoint for svxdr. It applies config from // svxdr.json over the defaults, starts the crawl/lastheard/federation/xlink // goroutines and the HTTP server, and returns when ctx is cancelled. func Run(rctx context.Context, c *config.Config) error { applyConfig(c) rdb := redis.NewClient(&redis.Options{ Network: "unix", Addr: redisSock, Password: redisPass, }) if err := rdb.Ping(ctx).Err(); err != nil { return fmt.Errorf("nodelist redis ping failed: %v (sock=%s)", err, redisSock) } // Restore the persisted via->node map (bridge login -> AllStar node) so the // web's per-bridge "connected nodes" button works immediately after a // restart, before any fresh talker meta arrives. if vn, err := rdb.HGetAll(ctx, "nodelist:vianode").Result(); err == nil && len(vn) > 0 { live.mu.Lock() for via, v := range vn { node, system, _ := strings.Cut(v, "|") // stored as "node|system" live.viaNode[strings.ToUpper(via)] = bridgeInfo{Node: node, System: system} } live.mu.Unlock() log.Printf("restored %d via->node bridge mappings", len(vn)) } // Location column render config now comes from svxdr.json (lastheard formats // + reflector names derived from nodes[]), not a separate lastheard.json. applyRenderFromConfig(c) go poller(rdb) go logTailer(rdb, reflectorLogPath) xlinkStart(rdb) if mqttBroker != "" && mqttPrefix != "" { go startMqtt() } else { log.Printf("mqtt disabled (set MQTT_BROKER and MQTT_TOPIC_PREFIX to enable)") } // Topology: crawl the network from our uplink (derived from svxreflector.conf) // so the graph shows every reflector AND the trunk links between them. Each // reflector's OWN /status carries the real peer hosts that draw the edges — an // aggregated hub status does not, so nothing else can reproduce the links. // Runs on a gentle interval (EXTERNAL_PULL_INTERVAL, default 60s) with last-good // retention, so it is spread out and never hammers any single server. go externalPuller() // Canonical TG->name list (svx-web /tglist.php). No-op until tglist.url is set // in config; when set, it feeds systemForTG so Location renders the same name // on every box that points at the same list. go tglistPuller(rctx, c) mux := http.NewServeMux() mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/plain; charset=utf-8") w.WriteHeader(200) _, _ = w.Write([]byte("ok\n")) }) mux.HandleFunc("/api/status", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json; charset=utf-8") b, err := rdb.Get(ctx, statusKey).Bytes() if err != nil { http.Error(w, `{"msg":"no status yet"}`, http.StatusServiceUnavailable) return } writeMaybeGzip(w, r, b) }) // Fas 2: federation gap report — which reflectors/callsigns the network-wide // aggregate has that our own crawl does not cover yet. Diagnostic only. mux.HandleFunc("/api/gaps", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json; charset=utf-8") federationGap.mu.RLock() out := map[string]any{ "updatedAt": federationGap.updatedAt, "aggregateURL": federationAggURL, "aggregateTotal": federationGap.aggTotal, "weCover": federationGap.haveTotal, "missingCallsigns": federationGap.missingCS, "missingReflectors": federationGap.missingRefs, "error": federationGap.err, } federationGap.mu.RUnlock() b, _ := json.Marshal(out) writeMaybeGzip(w, r, b) }) // This box's OWN bridges (via -> {node, system}), for other reflectors to // crawl and union into their network-wide bridge directory. Returns only // the LOCAL map (never the crawled union) so crawling can't amplify/loop. mux.HandleFunc("/api/bridge-nodes", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json; charset=utf-8") out := map[string]bridgeInfo{} live.mu.RLock() for via, bi := range live.viaNode { out[via] = bi } live.mu.RUnlock() b, _ := json.Marshal(out) writeMaybeGzip(w, r, b) }) // The complete network structure on its own, for other systems to fetch. mux.HandleFunc("/api/network", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json; charset=utf-8") networkMu.RLock() b := networkJSON networkMu.RUnlock() writeMaybeGzip(w, r, b) }) mux.HandleFunc("/api/stream", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream; charset=utf-8") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") flusher, ok := w.(http.Flusher) if !ok { http.Error(w, "SSE not supported", 500) return } sub := rdb.Subscribe(ctx, pubsubChan) defer sub.Close() _, _ = io.WriteString(w, "event: update\ndata: 1\n\n") flusher.Flush() pingT := time.NewTicker(25 * time.Second) defer pingT.Stop() ch := sub.Channel() for { select { case <-r.Context().Done(): return case <-pingT.C: _, _ = io.WriteString(w, "event: ping\ndata: 1\n\n") flusher.Flush() case <-ch: _, _ = io.WriteString(w, "event: update\ndata: 1\n\n") flusher.Flush() } } }) mux.HandleFunc("/api/debug", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json; charset=utf-8") resp := map[string]any{ "listen": listenAddr, "reflectorURL": reflectorURL, "pollInterval": pollInterval.String(), "redisSock": redisSock, "statusKey": statusKey, "nodePrefix": nodeKeyPrefix, "onlineTTL": effectiveOnlineTTL().String(), "debug": debug, } // Current talker meta (MQTT or ingest fed), with age so the freshness // gating can be judged without knowing the TTL. type metaDump struct { talkerMeta AgeSec int64 `json:"age_s"` } metas := map[string]metaDump{} live.mu.RLock() for tg, m := range live.tgMeta { metas[strconv.Itoa(tg)] = metaDump{m, int64(time.Since(m.At).Seconds())} } live.mu.RUnlock() resp["talkerMeta"] = metas st.mu.RLock() resp["startedAt"] = st.StartedAt.Format(time.RFC3339) resp["lastFetchOKAt"] = tOrEmpty(st.LastFetchOKAt) resp["lastFetchURL"] = st.LastFetchURL resp["lastFetchBytes"] = st.LastFetchBytes resp["lastFetchErr"] = st.LastFetchErr resp["lastParsedNodes"] = st.LastParsedNodes resp["lastStatusHash"] = st.LastStatusHash resp["lastRedisOKAt"] = tOrEmpty(st.LastRedisOKAt) resp["lastRedisErr"] = st.LastRedisErr st.mu.RUnlock() externalState.mu.RLock() resp["externalPullURLs"] = externalPullURLs() // derived from conf (or override) resp["externalConf"] = svxreflectorConf resp["externalSources"] = len(externalState.sources) resp["externalGenerated"] = externalState.generated resp["externalLastOKAt"] = tOrEmpty(externalState.lastOKAt) resp["externalLastErr"] = externalState.lastErr externalState.mu.RUnlock() _ = json.NewEncoder(w).Encode(resp) }) mux.HandleFunc("/api/lastheard", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json; charset=utf-8") limit := 200 if s := r.URL.Query().Get("limit"); s != "" { var n int if _, err := fmt.Sscanf(s, "%d", &n); err == nil && n > 0 && n <= 1000 { limit = n } } now := time.Now().Unix() minTs := now - lastheardWindowSec vals, err := rdb.ZRevRangeByScore(ctx, lastheardKey, &redis.ZRangeBy{ Min: fmt.Sprintf("%d", minTs), Max: "+inf", Offset: 0, Count: int64(limit), }).Result() if err != nil { http.Error(w, `{"msg":"redis error"}`, 500) return } out := make([]json.RawMessage, 0, len(vals)) for _, v := range vals { out = append(out, json.RawMessage(v)) } _ = json.NewEncoder(w).Encode(map[string]any{ "window_s": lastheardWindowSec, "count": len(out), "events": out, }) }) mux.HandleFunc("/api/nodekeys", func(w http.ResponseWriter, r *http.Request) { call := strings.TrimSpace(r.URL.Query().Get("call")) if call == "" { http.Error(w, `{"msg":"missing ?call="}`, 400) return } keys := []string{ statusHashKey, statusKey, fmt.Sprintf("%s/%s/online", nodeKeyPrefix, call), fmt.Sprintf("%s/%s/hash", nodeKeyPrefix, call), fmt.Sprintf("%s/%s/json", nodeKeyPrefix, call), fmt.Sprintf("%s/%s/rx", nodeKeyPrefix, call), fmt.Sprintf("%s/%s/tx", nodeKeyPrefix, call), fmt.Sprintf("%s/%s/toneToTalkgroup", nodeKeyPrefix, call), heartbeatKey, } out := map[string]any{"call": call, "keys": map[string]any{}} m := out["keys"].(map[string]any) for _, k := range keys { v, err := rdb.Get(ctx, k).Result() if err != nil { m[k] = nil } else { m[k] = v } } w.Header().Set("Content-Type", "application/json; charset=utf-8") _ = json.NewEncoder(w).Encode(out) }) // Talker metadata pushed from the USRP bridge's svxlink TCL hook // (usrp_stationdata_received) over HTTPS. Feeds the same live.tgMeta // store as the MQTT talker//meta leg, so via-gating and TTL apply // unchanged. A bare numeric id is resolved to a callsign via the // radioid tables callmgr2 keeps in redis (DMR first, then NXDN). mux.HandleFunc("/api/talker-meta", func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, `{"msg":"POST only"}`, http.StatusMethodNotAllowed) return } // Zero-config: en process på SAMMA burk (callmgr-satelliten, lokal astmeta) // får POSTa utan token — så en DMR/SVX-brygga bara funkar utan konfig. // Fjärranrop kräver X-Ingest-Token som förr (och att ett token är satt). if !isLoopbackAddr(r.RemoteAddr) { if ingestToken == "" { http.Error(w, `{"msg":"ingest disabled"}`, http.StatusNotFound) return } tok := r.Header.Get("X-Ingest-Token") if subtle.ConstantTimeCompare([]byte(tok), []byte(ingestToken)) != 1 { http.Error(w, `{"msg":"bad token"}`, http.StatusUnauthorized) return } } var in struct { TG int `json:"tg"` Event string `json:"event"` Callsign string `json:"callsign"` Node string `json:"node"` System string `json:"system"` Via string `json:"via"` Network string `json:"network"` // ingress mode (AllStar / DMR / DMR/YSF …) } body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 4096)) if err != nil || json.Unmarshal(body, &in) != nil || in.TG <= 0 { http.Error(w, `{"msg":"bad request"}`, http.StatusBadRequest) return } w.Header().Set("Content-Type", "application/json; charset=utf-8") if in.Event == "stop" { // Intentionally no clear: sub-second re-keys would race it // (same design as astmeta). The TTL ages the entry out. _, _ = w.Write([]byte(`{"ok":true}`)) return } cs := strings.TrimSpace(in.Callsign) node := strings.TrimSpace(in.Node) if (cs == "" || cs == "N0CALL" || isDigits(cs)) && isDigits(node) { for _, k := range []string{ "/registry/ext/dmr/db/users/id/" + node, "/registry/ext/nxdn/db/users/id/" + node, } { if v, err := rdb.HGet(ctx, k, "callsign").Result(); err == nil && v != "" { cs = v break } } } if cs == "" { http.Error(w, `{"msg":"no callsign"}`, http.StatusBadRequest) return } m := talkerMeta{Callsign: cs, Node: node, System: strings.TrimSpace(in.System), Via: strings.TrimSpace(in.Via), Network: strings.TrimSpace(in.Network), At: time.Now()} live.mu.Lock() live.tgMeta[in.TG] = m // Track the station for /api/allstar-nodes and prune expired entries. live.asNodes[cs] = allstarNode{Callsign: cs, Node: node, System: m.System, Via: m.Via, TG: in.TG, At: m.At} for k, v := range live.asNodes { if time.Since(v.At) > allstarNodesTTL { delete(live.asNodes, k) } } // Long-lived via->node map: which AllStar node a bridge login speaks // through, so the web can offer a "connected nodes" button on that // bridge row even when idle (no fresh talker meta). Derived, never // hard-coded, and persisted so it survives a restart. if m.Via != "" && node != "" { via := strings.ToUpper(m.Via) bi := bridgeInfo{Node: node, System: m.System} if live.viaNode[via] != bi { live.viaNode[via] = bi if rdb != nil { rdb.HSet(ctx, "nodelist:vianode", via, node+"|"+m.System) } } } live.mu.Unlock() if debug { log.Printf("DEBUG ingest talker meta tg=%d callsign=%q node=%q via=%q", in.TG, m.Callsign, m.Node, m.Via) } _, _ = w.Write([]byte(`{"ok":true}`)) }) // The bridge-heard stations as a reflector-status style node list: // every station seen via the talker-meta ingest within // ALLSTAR_NODES_TTL, keyed by callsign. Location comes from the // radioid tables in redis when the station's id is known. mux.HandleFunc("/api/allstar-nodes", func(w http.ResponseWriter, r *http.Request) { type nodeOut struct { Hidden bool `json:"hidden"` IsTalker bool `json:"isTalker"` MonitoredTGs []int `json:"monitoredTGs"` NodeClass string `json:"nodeClass"` NodeLocation string `json:"nodeLocation"` RestrictedTG bool `json:"restrictedTG"` SW string `json:"sw"` Sysop string `json:"sysop"` TG int `json:"tg"` Node string `json:"node,omitempty"` Via string `json:"via,omitempty"` LastHeard int64 `json:"last_heard"` } out := map[string]nodeOut{} now := time.Now() live.mu.RLock() for cs, n := range live.asNodes { if now.Sub(n.At) > allstarNodesTTL { continue } m, ok := live.tgMeta[n.TG] isTalker := ok && m.Callsign == cs && now.Sub(m.At) <= talkerMetaTTL loc := "via " + n.Via if n.Via == "" { loc = n.System } out[cs] = nodeOut{IsTalker: isTalker, MonitoredTGs: []int{n.TG}, NodeClass: "AllStar", NodeLocation: loc, SW: "AllStarLink", Sysop: cs, TG: n.TG, Node: n.Node, Via: n.Via, LastHeard: n.At.Unix()} } live.mu.RUnlock() // Location from the radioid tables — outside the lock (redis I/O). for cs, n := range out { if !isDigits(n.Node) { continue } for _, k := range []string{ "/registry/ext/dmr/db/users/id/" + n.Node, "/registry/ext/nxdn/db/users/id/" + n.Node, } { vals, err := rdb.HMGet(ctx, k, "city", "state", "country").Result() if err != nil { continue } parts := []string{} for _, v := range vals { if s, ok := v.(string); ok && s != "" { parts = append(parts, s) } } if len(parts) > 0 { n.NodeLocation = strings.Join(parts, ", ") out[cs] = n break } } } resp := map[string]any{ "mode": "bridge", "ttl_s": int(allstarNodesTTL.Seconds()), "nodes": out, } w.Header().Set("Content-Type", "application/json; charset=utf-8") _ = json.NewEncoder(w).Encode(resp) }) // /api/remote-nodes?node= — ADDITIVE read-only endpoint: the nodes // currently linked to an AllStar node, from stats.allstarlink.org, reshaped // to {node, callsign, location, keyed}. Server-side fetch, short cache. Does // not touch any existing response shape, so all current readers are safe. mux.HandleFunc("/api/remote-nodes", func(w http.ResponseWriter, r *http.Request) { node := strings.TrimSpace(r.URL.Query().Get("node")) if !isDigits(node) { http.Error(w, `{"msg":"node query param (digits) required"}`, http.StatusBadRequest) return } resp, err := remoteNodes(node) if err != nil { http.Error(w, `{"msg":"upstream fetch failed"}`, http.StatusBadGateway) return } w.Header().Set("Content-Type", "application/json; charset=utf-8") _ = json.NewEncoder(w).Encode(resp) }) // All-Nodes flat layer: /api/nodes (full) + /api/nodes-stream (per-node deltas). nodeFlatStart(mux, rdb) // Monitor Nodes (single-endpoint): /api/monitor-nodes — Local + per-trunk. monitorNodesStart(mux) // Monitor Trunks: /api/local-trunks — ONLY the trunks this reflector is // configured to connect to, read from the local svxreflector.conf. localTrunksStart(mux) srv := &http.Server{ Addr: listenAddr, Handler: securityHeaders(mux), ReadHeaderTimeout: 5 * time.Second, } log.Printf("nodelist listen=%s reflector=%s poll=%s redis=unix:%s statusKey=%s nodePrefix=%s onlineTTL=%s debug=%v", listenAddr, reflectorURL, pollInterval, redisSock, statusKey, nodeKeyPrefix, effectiveOnlineTTL(), debug) go func() { <-rctx.Done() sc, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() _ = srv.Shutdown(sc) }() if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { return err } return nil } func tOrEmpty(t time.Time) string { if t.IsZero() { return "" } return t.Format(time.RFC3339) } func effectiveOnlineTTL() time.Duration { onlineTTL := nodeOnlineTTL if onlineTTL <= 0 { onlineTTL = nodeTTL } if onlineTTL <= 0 { onlineTTL = maxDur(15*time.Second, 3*pollInterval) } return onlineTTL } // ---- /api/remote-nodes: AllStarLink stats proxy (additive, read-only) -------- var ( remoteHTTP = &http.Client{Timeout: 8 * time.Second} remoteMu sync.Mutex remoteCache = map[string]remoteEntry{} ) const remoteTTL = 8 * time.Second type remoteEntry struct { at time.Time resp *remoteResp } type remoteNode struct { Node string `json:"node"` Callsign string `json:"callsign"` Location string `json:"location"` Keyed bool `json:"keyed"` } type remoteResp struct { Node string `json:"node"` // the AllStar node we queried Count int `json:"count"` Nodes []remoteNode `json:"nodes"` FetchedAt int64 `json:"fetched_at"` } // asStats mirrors only the fields we consume from stats.allstarlink.org. // The connected nodes are node-number strings under stats.data.links; this // endpoint carries no callsign/location (resolved separately, TODO astdb). type asStats struct { Stats struct { Data struct { Links []string `json:"links"` } `json:"data"` } `json:"stats"` } // remoteNodes fetches (and briefly caches) the linked-node list for an AllStar // node from the public stats API, reshaped for the UI. func remoteNodes(node string) (*remoteResp, error) { remoteMu.Lock() if e, ok := remoteCache[node]; ok && time.Since(e.at) < remoteTTL { remoteMu.Unlock() return e.resp, nil } remoteMu.Unlock() req, err := http.NewRequest(http.MethodGet, "https://stats.allstarlink.org/api/stats/"+node, nil) if err != nil { return nil, err } req.Header.Set("Accept", "application/json") hr, err := remoteHTTP.Do(req) if err != nil { return nil, err } defer hr.Body.Close() if hr.StatusCode >= 300 { return nil, fmt.Errorf("stats HTTP %d", hr.StatusCode) } var s asStats if err := json.NewDecoder(io.LimitReader(hr.Body, 8<<20)).Decode(&s); err != nil { return nil, err } ensureAstdb() // background-loads the node->callsign db if stale (non-blocking) resp := &remoteResp{Node: node, FetchedAt: time.Now().Unix()} for _, ln := range s.Stats.Data.Links { n := strings.TrimSpace(ln) if n == "" { continue } cs, loc := astdbLookup(n) resp.Nodes = append(resp.Nodes, remoteNode{Node: n, Callsign: cs, Location: loc}) } resp.Count = len(resp.Nodes) remoteMu.Lock() remoteCache[node] = remoteEntry{at: time.Now(), resp: resp} remoteMu.Unlock() return resp, nil } // ---- astdb: AllStar node -> callsign/location (background-loaded, cached) ---- var ( astdbURL = env("ASTDB_URL", "http://allmondb.allstarlink.org/") // node|call|desc|loc astdbLine = regexp.MustCompile(`^([0-9]+)\|([^|]*)\|([^|]*)\|(.*)$`) astdbMu sync.RWMutex astdbMap = map[string][2]string{} // node -> {callsign, location} astdbAt time.Time astdbBusy bool ) const astdbTTL = 6 * time.Hour // ensureAstdb starts a background reload when the cache is stale. Non-blocking: // the caller uses whatever is currently loaded (empty until the first load // finishes), so a request never waits on the (large) allmondb download. func ensureAstdb() { astdbMu.Lock() stale := time.Since(astdbAt) >= astdbTTL if astdbBusy || !stale { astdbMu.Unlock() return } astdbBusy = true astdbMu.Unlock() go loadAstdb() } func loadAstdb() { defer func() { astdbMu.Lock(); astdbBusy = false; astdbMu.Unlock() }() if astdbURL == "" { return } hr, err := remoteHTTP.Get(astdbURL) if err != nil { log.Printf("astdb: fetch failed: %v", err) return } defer hr.Body.Close() m := map[string][2]string{} sc := bufio.NewScanner(io.LimitReader(hr.Body, 64<<20)) sc.Buffer(make([]byte, 1<<20), 1<<20) for sc.Scan() { if f := astdbLine.FindStringSubmatch(sc.Text()); f != nil { m[f[1]] = [2]string{strings.TrimSpace(f[2]), strings.TrimSpace(f[4])} } } if len(m) > 0 { astdbMu.Lock() astdbMap = m astdbAt = time.Now() astdbMu.Unlock() log.Printf("astdb: loaded %d nodes", len(m)) } } func astdbLookup(node string) (callsign, location string) { astdbMu.RLock() defer astdbMu.RUnlock() if v, ok := astdbMap[node]; ok { return v[0], v[1] } return "", "" } func maxDur(a, b time.Duration) time.Duration { if a > b { return a } return b } func poller(rdb *redis.Client) { defer func() { if r := recover(); r != nil { log.Printf("poller panic: %v (will NOT recover)", r) os.Exit(2) } }() t := time.NewTicker(pollInterval) defer t.Stop() onlineTTL := effectiveOnlineTTL() nodeOnlineKey := func(cs string) string { return fmt.Sprintf("%s/%s/online", nodeKeyPrefix, cs) } prevOnline := map[string]struct{}{} if arr, err := rdb.SMembers(ctx, nodesOnlineKey).Result(); err == nil { for _, cs := range arr { prevOnline[cs] = struct{}{} } } var lastFetchErrLog time.Time for range t.C { body, usedURL, err := fetchReflectorStatus(reflectorURL) if err != nil { st.mu.Lock() st.LastFetchErr = err.Error() st.mu.Unlock() if debug { log.Printf("DEBUG fetch failed: %v (url=%s)", err, reflectorURL) } if time.Since(lastFetchErrLog) > 30*time.Second { log.Printf("reflector fetch failed: %v (url=%s)", err, reflectorURL) lastFetchErrLog = time.Now() } continue } st.mu.Lock() st.LastFetchOKAt = time.Now() st.LastFetchURL = usedURL st.LastFetchBytes = len(body) st.LastFetchErr = "" st.mu.Unlock() var env statusEnvelope if err := json.Unmarshal(body, &env); err != nil { st.mu.Lock() st.LastFetchErr = "parse: " + err.Error() st.mu.Unlock() log.Printf("status json parse failed: %v (url=%s)", err, usedURL) continue } if env.Nodes == nil { env.Nodes = map[string]json.RawMessage{} } st.mu.Lock() st.LastParsedNodes = len(env.Nodes) st.mu.Unlock() // Refresh the real-callsign lookup maps for the log tailer. newTgTalker := map[int]string{} newTgTalkerAt := map[int]int64{} newTgLoc := map[int]string{} newTgMode := map[int]string{} crawlNow := time.Now().Unix() for _, tr := range env.Trunks { for tgs, cs := range tr.ActiveTalkers { if cs == "" { continue } if tg, e := strconv.Atoi(tgs); e == nil && tg > 0 { newTgTalker[tg] = cs newTgTalkerAt[tg] = crawlNow } } for _, n := range tr.Nodes { if n.IsTalker && n.Callsign != "" && n.TG > 0 { newTgTalker[n.TG] = n.Callsign newTgTalkerAt[n.TG] = crawlNow } } } // Satellite trunks (chan_svx_sat / chan_pbx_svx) carry their talkers in a // separate section, not env.Trunks. The node's callsign is the real caller // (SA7AUX); its gateway label rides in nodeLocation (SA7AUX-PBX) -> the // Location column, like an AllStar node's node_info.json. for _, sat := range env.Satellites { for _, n := range sat.Nodes { if !n.IsTalker || n.Callsign == "" || n.TG <= 0 { continue } newTgTalker[n.TG] = n.Callsign newTgTalkerAt[n.TG] = crawlNow if n.NodeLocation != "" { newTgLoc[n.TG] = n.NodeLocation } if m := monModeOf(n.SW); m != "" { newTgMode[n.TG] = m } } } newNodeSysop := map[string]string{} for cs, raw := range env.Nodes { if sy := extractSysop(raw); sy != "" { newNodeSysop[cs] = sy } } // Sticky: a talker-start from the log tail can race ahead of this crawl, // so a TG the crawl momentarily lost (talker still active, roster lagging) // would drop back to the raw trunk name. Carry a very-recent value forward // (short 15s window) so resolveTalker keeps the real callsign across the gap. const stickyWindow = int64(15) live.mu.Lock() for tg, cs := range live.tgTalker { if _, ok := newTgTalker[tg]; ok { continue } if at := live.tgTalkerAt[tg]; at > 0 && crawlNow-at <= stickyWindow { newTgTalker[tg] = cs newTgTalkerAt[tg] = at } } live.tgTalker = newTgTalker live.tgTalkerAt = newTgTalkerAt live.tgLoc = newTgLoc live.tgMode = newTgMode live.nodeSysop = newNodeSysop live.mu.Unlock() nowOnline := map[string]struct{}{} for cs := range env.Nodes { nowOnline[cs] = struct{}{} } added, removed := diffSets(prevOnline, nowOnline) pipe := rdb.Pipeline() pipe.Set(ctx, heartbeatKey, fmt.Sprint(time.Now().Unix()), 5*time.Second) // Merge MQTT-sourced trunk nodes into the served status so the // dashboard's node views show trunk-side nodes + their isTalker state. storeBody := body if merged := mergeMqttNodes(env.Nodes); len(merged) > len(env.Nodes) { var top map[string]json.RawMessage if json.Unmarshal(body, &top) == nil { if nb, err := json.Marshal(merged); err == nil { top["nodes"] = nb if bb, err := json.Marshal(top); err == nil { storeBody = bb } } } } // Inject the network-wide picture (ext_sources + network_trunks) so the // Monitor views can show every reflector's nodes and all trunks. storeBody = injectExternal(storeBody) // Fresh bridge talker meta for the "Talking now" strip. storeBody = injectTalkerMeta(storeBody) h := hash(storeBody) pipe.Set(ctx, statusKey, storeBody, 0) pipe.Set(ctx, statusHashKey, h, 0) pipe.Publish(ctx, pubsubChan, "1") st.mu.Lock() st.LastStatusHash = h st.mu.Unlock() if len(added) > 0 { members := make([]interface{}, 0, len(added)) for _, cs := range added { members = append(members, cs) } pipe.SAdd(ctx, nodesOnlineKey, members...) } if len(removed) > 0 { members := make([]interface{}, 0, len(removed)) for _, cs := range removed { members = append(members, cs) } pipe.SRem(ctx, nodesOnlineKey, members...) } for cs, raw := range env.Nodes { pipe.Set(ctx, nodeOnlineKey(cs), "1", onlineTTL) nh := hash(raw) pipe.Set(ctx, fmt.Sprintf("%s/%s/hash", nodeKeyPrefix, cs), nh, 0) // IMPORTANT: go-redis can't marshal json.RawMessage directly; cast to []byte. pipe.Set(ctx, fmt.Sprintf("%s/%s/json", nodeKeyPrefix, cs), []byte(raw), 0) rx, tx := extractRxTx(raw) pipe.Set(ctx, fmt.Sprintf("%s/%s/rx", nodeKeyPrefix, cs), fmt.Sprintf("%.5f", rx), 0) pipe.Set(ctx, fmt.Sprintf("%s/%s/tx", nodeKeyPrefix, cs), fmt.Sprintf("%.5f", tx), 0) tones := extractToneToTG(raw) if tones != nil && len(tones) > 0 { b, _ := json.Marshal(tones) pipe.Set(ctx, fmt.Sprintf("%s/%s/toneToTalkgroup", nodeKeyPrefix, cs), b, 0) } else { pipe.Del(ctx, fmt.Sprintf("%s/%s/toneToTalkgroup", nodeKeyPrefix, cs)) } } for _, cs := range added { pipe.Publish(ctx, nodeConnectChan, cs) pipe.Publish(ctx, nodeUpdateChan, "connect:"+cs) } for _, cs := range removed { pipe.Publish(ctx, nodeDisconnectChan, cs) pipe.Publish(ctx, nodeUpdateChan, "disconnect:"+cs) pipe.Del(ctx, nodeOnlineKey(cs)) } // lastheard events come from the reflector-log tailer (logTailer goroutine) if _, execErr := pipe.Exec(ctx); execErr != nil { st.mu.Lock() st.LastRedisErr = execErr.Error() st.mu.Unlock() log.Printf("redis exec failed: %v", execErr) continue } st.mu.Lock() st.LastRedisOKAt = time.Now() st.LastRedisErr = "" st.mu.Unlock() if debug { log.Printf("DEBUG wrote: nodes=%d added=%d removed=%d", len(env.Nodes), len(added), len(removed)) } prevOnline = nowOnline } } func extractRxTx(raw json.RawMessage) (rx float64, tx float64) { var ni nodeInfo if err := json.Unmarshal(raw, &ni); err != nil { return 0, 0 } if ni.RxFreq > 0 { rx = ni.RxFreq } else if ni.RX > 0 { rx = ni.RX } if ni.TxFreq > 0 { tx = ni.TxFreq } else if ni.TX > 0 { tx = ni.TX } if (rx <= 0 || tx <= 0) && len(ni.QTH) > 0 { q := ni.QTH[0] if rx <= 0 { for _, v := range q.RX { if v.Freq > 0 { rx = v.Freq break } } } if tx <= 0 { for _, v := range q.TX { if v.Freq > 0 { tx = v.Freq break } } } } if rx > 0 && tx <= 0 { tx = rx } return } func extractToneToTG(raw json.RawMessage) map[string]int { var ni nodeInfo if err := json.Unmarshal(raw, &ni); err != nil { return nil } return ni.ToneToTalkgroup } func extractSysop(raw json.RawMessage) string { var ni nodeInfo if err := json.Unmarshal(raw, &ni); err != nil { return "" } return strings.TrimSpace(ni.Sysop) } var ( // Matches both a local/node talker ("... KT4ROY-A: Talker start on TG #N") // and a trunk/satellite-delivered one ("... [trunk] SA7AUX: Trunk talker // start on TG #N"). The optional "Trunk " group tags the source. talkerLineRe = regexp.MustCompile(`^(.+?): (Trunk )?[Tt]alker (start|stop) on TG #(\d+)\s*$`) logDateRe = regexp.MustCompile(`^(\w{3}\s+\d{1,2}\s+\w{3}\s+\d{4}\s+\d{2}:\d{2}:\d{2}\s+(?:AM|PM))\s+[A-Z]{2,5}$`) ) type talkerLogEvent struct { Ts int64 Callsign string TG int IsStop bool IsTrunk bool } func parseTalkerLogLine(line string) (ev talkerLogEvent, ok bool) { m := talkerLineRe.FindStringSubmatch(line) if m == nil { return } prefix := m[1] idx := strings.LastIndex(prefix, ": ") if idx < 0 { return } datePart := prefix[:idx] tail := prefix[idx+2:] ev.Ts = parseReflectorDate(datePart) if ev.Ts == 0 { ev.Ts = time.Now().Unix() } ev.IsTrunk = m[2] != "" || strings.Contains(tail, "-> Trunk") fields := strings.Fields(tail) // Newer reflector builds prefix log lines with severity/component tags // ("[info] [client] KT4ROY-AS: Talker start ..."); skip any such tags so // the callsign is found in both the old and the new log format. for len(fields) > 0 && strings.HasPrefix(fields[0], "[") && strings.HasSuffix(fields[0], "]") { fields = fields[1:] } if len(fields) == 0 { return } ev.Callsign = strings.TrimRight(fields[0], ":,") ev.IsStop = m[3] == "stop" tg, err := strconv.Atoi(m[4]) if err != nil { return } ev.TG = tg ok = true return } func parseReflectorDate(s string) int64 { s = strings.TrimSpace(s) m := logDateRe.FindStringSubmatch(s) if m == nil { t, err := time.ParseInLocation("Mon 2 Jan 2006 03:04:05 PM MST", s, time.Local) if err == nil { return t.Unix() } return 0 } t, err := time.ParseInLocation("Mon 2 Jan 2006 03:04:05 PM", m[1], time.Local) if err != nil { return 0 } return t.Unix() } type liveTalker struct { StartTs int64 TG int IsTrunk bool Callsign string // resolved real callsign, captured at talker-start Trunk string // node/trunk name when it differs from the callsign Node string // source-network node/id (AllStar node, DMR id), from meta System string // source system name (e.g. "East Coast Reflector"), from meta Via string // reflector callsign the bridge speaks through, from meta Network string // ingress mode: AllStar / SVXLink / trunk / DMR / DMR/YSF … LogName string // raw talker name from the reflector log (for via-match) } // lastheardRenderCfg controls how bridge/AllStar rows render in lastheard, // editable at runtime via a JSON file (hot-reloaded) — NO recompile, nothing // hardcoded. Path: LASTHEARD_RENDER_JSON env, default /etc/nodelist/lastheard.json. // {system}/{node}/{talker} in the format strings are filled dynamically. // reflector_names gives a friendly System name to reflectors astmeta doesn't // name (e.g. "ASL-dodropin" -> "Do Drop In"), keyed by the bridge/log name. type lastheardRenderCfg struct { BridgeLocationFormat string `json:"bridge_location_format"` BridgeLocationFormatNoSystem string `json:"bridge_location_format_no_system"` ReflectorNames map[string]string `json:"reflector_names"` } var defaultRenderCfg = lastheardRenderCfg{ BridgeLocationFormat: "{system} via {node} / {talker}", BridgeLocationFormatNoSystem: "via {node} / {talker}", ReflectorNames: map[string]string{}, } var ( renderCfgMu sync.RWMutex renderCfg = defaultRenderCfg ) func renderConfig() lastheardRenderCfg { renderCfgMu.RLock() defer renderCfgMu.RUnlock() return renderCfg } // loadRenderConfig merges the JSON file over the built-in defaults, so the // binary works with no file and a partial file only overrides what it sets. func loadRenderConfig(path string) { cfg := defaultRenderCfg cfg.ReflectorNames = map[string]string{} if b, err := os.ReadFile(path); err == nil { var fc lastheardRenderCfg if jerr := json.Unmarshal(b, &fc); jerr != nil { log.Printf("lastheard render json %s: %v (keeping current)", path, jerr) return } if strings.TrimSpace(fc.BridgeLocationFormat) != "" { cfg.BridgeLocationFormat = fc.BridgeLocationFormat } if strings.TrimSpace(fc.BridgeLocationFormatNoSystem) != "" { cfg.BridgeLocationFormatNoSystem = fc.BridgeLocationFormatNoSystem } for k, v := range fc.ReflectorNames { cfg.ReflectorNames[k] = v } } // nodes.json (shared bridge map) merged on top: it is the source of truth // for reflector names + Location formats. Absent file = no-op. applyNodesRenderConfig(&cfg, nodesJSONPath()) renderCfgMu.Lock() renderCfg = cfg renderCfgMu.Unlock() } // watchRenderConfig hot-reloads whenever lastheard.json OR nodes.json changes, // so editing the shared bridge map updates the Location column with no restart. func watchRenderConfig(path string) { var last, lastNodes time.Time for { reload := false if fi, err := os.Stat(path); err == nil && fi.ModTime() != last { last = fi.ModTime() reload = true } if fi, err := os.Stat(nodesJSONPath()); err == nil && fi.ModTime() != lastNodes { lastNodes = fi.ModTime() reload = true } if reload { loadRenderConfig(path) } time.Sleep(3 * time.Second) } } // bridgeIdentity splits a talker into the (Callsign, Location) shown in // lastheard, uniform for PBX and AllStar (approved layout). PBX / plain svxlink: // Callsign = lt.Callsign (sip-tg / real callsign), Location = locForTG // (SA7AUX-PBX / blank). AllStar/bridge (meta System known): the BRIDGE // (lt.Via, e.g. KT4ROY-AS) is the Callsign, and Location carries // " via / " (e.g. "East Coast Reflector via 45225 / W2ECR"). // The AllStar node number is dynamic from astmeta (lt.Node), never hardcoded. func bridgeIdentity(lt liveTalker, tg int) (callsign, location string) { callsign = lt.Callsign location = locForTG(tg) bridge := lt.Via if bridge == "" { bridge = lt.Trunk } // A bridge/AllStar talker has a bridge node (Via/Trunk) AND meta (Node or // System) — the reflector name may be absent (e.g. ASL-dodropin). Plain // svxlink / PBX carry no bridge node -> untouched. if bridge == "" || (lt.Node == "" && lt.System == "") { return callsign, location } // FROZEN DESIGN: Location is the tglist name ONLY (already set above via // locForTG). We no longer build " via / " from live // astmeta — the tglist is the sole source, so the row reads identically on // every box. The bridge is still the Callsign (who spoke). callsign = bridge return callsign, location } func writeLastheardEvent(rdb *redis.Client, ev lastheardEvent) { b, err := json.Marshal(ev) if err != nil { return } pipe := rdb.Pipeline() pipe.ZAdd(ctx, lastheardKey, redis.Z{Score: float64(ev.Ts), Member: string(b)}) cutoff := time.Now().Unix() - lastheardWindowSec pipe.ZRemRangeByScore(ctx, lastheardKey, "-inf", fmt.Sprintf("(%d", cutoff)) if _, err := pipe.Exec(ctx); err != nil { log.Printf("lastheard write failed: %v", err) } } func sourceLabel(isTrunk bool) string { if isTrunk { return "trunk" } return "local" } // defaultNetwork is the ingress mode when no talker meta enriches the row: // a trunk-delivered talker enters over a reflector↔reflector trunk, everything // else is a plain svxlink client. A bridge (AllStar via astmeta, later DMR via // callmgr) overrides this with its own network string from the talker meta. func defaultNetwork(isTrunk bool) string { if isTrunk { return "trunk" } return "SVXLink" } func logTailer(rdb *redis.Client, path string) { defer func() { if r := recover(); r != nil { log.Printf("logTailer panic: %v", r) } }() active := map[string]liveTalker{} // key: callsign|tg keyOf := func(cs string, tg int) string { return fmt.Sprintf("%s|%d", cs, tg) } mkLive := func(ev talkerLogEvent) liveTalker { cs, tn := resolveTalker(ev.Callsign, ev.TG, ev.IsTrunk) lt := liveTalker{StartTs: ev.Ts, TG: ev.TG, IsTrunk: ev.IsTrunk, Callsign: cs, Trunk: tn, LogName: ev.Callsign, Network: defaultNetwork(ev.IsTrunk)} // Meta applies regardless of source (locally bridged traffic here, // trunk-delivered on the partner reflector) — gated by via-match. if m, ok := talkerMetaFor(ev.TG, ev.Callsign); ok { lt.Callsign, lt.Node, lt.System, lt.Via = m.Callsign, m.Node, m.System, m.Via if m.Network != "" { lt.Network = m.Network } if lt.Trunk == "" && ev.Callsign != m.Callsign { lt.Trunk = ev.Callsign } } // Gateway-node Backend (PBX etc.) wins over the generic "trunk" ingress. if md := modeForTG(ev.TG); md != "" { lt.Network = md } return lt } emit := func(start liveTalker, tg int, stopTs int64) { // The meta often arrives just after talker-start (astmeta polls AMI // about once a second), so short keyups miss it in mkLive. Re-check // at stop — the meta is still fresh at this point — so those rows // get the real source identity instead of the sysop fallback. if start.Node == "" { if m, ok := talkerMetaFor(tg, start.LogName); ok { if start.Trunk == "" && start.Callsign != m.Callsign { start.Trunk = start.Callsign } start.Callsign, start.Node, start.System, start.Via = m.Callsign, m.Node, m.System, m.Via if m.Network != "" { start.Network = m.Network } } } dur := stopTs - start.StartTs if dur < 1 { dur = 1 } // Re-apply the gateway Backend at stop: tgMode may not have been populated // yet when mkLive ran at talker-start (crawl lag on short keyups). if md := modeForTG(tg); md != "" { start.Network = md } // Location column, symmetric for PBX and AllStar: the gateway/reflector // descriptor lives here, the station stays in Callsign. PBX gets its // gateway nodeLocation via locForTG (e.g. SA7AUX-PBX); a bridge/AllStar // talker (no TG->location map) gets " via " (e.g. // "East Coast Reflector via KT4ROY-AS") so it renders like the PBX rows. // Callsign + Location split (uniform PBX/AllStar). See bridgeIdentity. csOut, loc := bridgeIdentity(start, tg) writeLastheardEvent(rdb, lastheardEvent{ Ts: start.StartTs, Callsign: csOut, Trunk: start.Trunk, TG: tg, Duration: dur, Location: loc, Source: sourceLabel(start.IsTrunk), Node: start.Node, System: start.System, Network: start.Network, }) } // 1) Backfill: scan whole file, replay events into ZSET (last 24h only). now0 := time.Now().Unix() // Timestamps already persisted in the lastheard set, so a restart's backfill // does not re-emit (duplicate) rows the previous run already wrote live — a // live row may be enriched ("X via X-PBX") while the raw backfill cannot // enrich (tgTalker is empty for historical events), producing a second row. existing := map[int64]bool{} if zs, zerr := rdb.ZRangeWithScores(ctx, lastheardKey, 0, -1).Result(); zerr == nil { for _, z := range zs { existing[int64(z.Score)] = true } } if f, err := os.Open(path); err == nil { sc := bufio.NewScanner(f) sc.Buffer(make([]byte, 64*1024), 1024*1024) for sc.Scan() { ev, ok := parseTalkerLogLine(sc.Text()) if !ok { continue } if ev.Ts < now0-lastheardWindowSec { if !ev.IsStop { continue } delete(active, keyOf(ev.Callsign, ev.TG)) continue } k := keyOf(ev.Callsign, ev.TG) if !ev.IsStop { active[k] = mkLive(ev) } else if st, was := active[k]; was { if !existing[st.StartTs] { // already persisted by a previous run → don't duplicate emit(st, ev.TG, ev.Ts) } delete(active, k) } } f.Close() log.Printf("logTailer: backfill done, %d in-flight talkers", len(active)) } else { log.Printf("logTailer: cannot open %s for backfill: %v", path, err) } // 2) Tail mode: follow new lines, handle rotation. for { f, err := os.Open(path) if err != nil { time.Sleep(2 * time.Second) continue } if _, err := f.Seek(0, io.SeekEnd); err != nil { f.Close() time.Sleep(2 * time.Second) continue } fi0, _ := f.Stat() reader := bufio.NewReader(f) for { line, err := reader.ReadString('\n') if err == nil { line = strings.TrimRight(line, "\r\n") if ev, ok := parseTalkerLogLine(line); ok { k := keyOf(ev.Callsign, ev.TG) if !ev.IsStop { lt := mkLive(ev) active[k] = lt // Live only (not backfill): announce the SVX talker to the mesh. xlinkPublishTalker(lt.Callsign, lt.System, lt.Trunk, lt.Node, ev.TG, lt.Network, "start", ev.Ts) } else if st, was := active[k]; was { xlinkPublishTalker(st.Callsign, st.System, st.Trunk, st.Node, ev.TG, st.Network, "stop", ev.Ts) emit(st, ev.TG, ev.Ts) delete(active, k) } else { lt := mkLive(ev) cs3, loc3 := bridgeIdentity(lt, ev.TG) writeLastheardEvent(rdb, lastheardEvent{ Ts: ev.Ts, Callsign: cs3, Trunk: lt.Trunk, TG: ev.TG, Duration: 1, Location: loc3, Source: sourceLabel(ev.IsTrunk), Node: lt.Node, System: lt.System, Network: lt.Network, }) } } continue } if err == io.EOF { time.Sleep(300 * time.Millisecond) fi1, sErr := os.Stat(path) if sErr == nil && fi0 != nil { if !os.SameFile(fi0, fi1) || fi1.Size() < fi0.Size() { f.Close() break } } continue } log.Printf("logTailer read error: %v", err) f.Close() break } time.Sleep(500 * time.Millisecond) } } func extractTalkerInfo(raw json.RawMessage) (talking bool, tg int, location, rxName string, siglev float64) { var ni nodeInfo if err := json.Unmarshal(raw, &ni); err != nil { return } talking = ni.IsTalker tg = ni.TG location = ni.NodeLocation if len(ni.QTH) > 0 { for _, q := range ni.QTH { if location == "" && q.Name != "" { location = q.Name } for _, rx := range q.RX { if rx.Active { if rx.Name != "" { rxName = rx.Name if q.Name != "" { rxName = rx.Name + " " + q.Name } } siglev = rx.Siglev return } } } } return } func diffSets(prev, now map[string]struct{}) (added []string, removed []string) { for k := range now { if _, ok := prev[k]; !ok { added = append(added, k) } } for k := range prev { if _, ok := now[k]; !ok { removed = append(removed, k) } } sort.Strings(added) sort.Strings(removed) return } func fetchReflectorStatus(url string) ([]byte, string, error) { try := func(u string) ([]byte, string, error) { b, err := fetch(u) if err != nil { return nil, u, err } if !json.Valid(b) { return nil, u, fmt.Errorf("not json (http 200)") } var m map[string]json.RawMessage if err := json.Unmarshal(b, &m); err != nil { return nil, u, err } if _, ok := m["nodes"]; !ok { return nil, u, fmt.Errorf("missing top-level 'nodes'") } return b, u, nil } body, used, err := try(url) if err == nil { return body, used, nil } alt := altStatusURL(url) if alt == "" || alt == url { return nil, used, err } return try(alt) } func altStatusURL(url string) string { if strings.HasSuffix(url, "/status") { return url + "/json" } if strings.HasSuffix(url, "/status/json") { return strings.TrimSuffix(url, "/json") } if strings.HasSuffix(url, "/status/") { return strings.TrimSuffix(url, "/") + "/json" } if strings.HasSuffix(url, "/status/json/") { alt := strings.TrimSuffix(url, "/") alt = strings.TrimSuffix(alt, "/json") return alt } return "" } func fetch(url string) ([]byte, error) { c := &http.Client{Timeout: 2 * time.Second} resp, err := c.Get(url) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode != 200 { _, _ = io.Copy(io.Discard, resp.Body) return nil, fmt.Errorf("http %d", resp.StatusCode) } return io.ReadAll(resp.Body) } func hash(b []byte) string { sum := sha256.Sum256(b) return hex.EncodeToString(sum[:]) } func securityHeaders(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("X-Content-Type-Options", "nosniff") w.Header().Set("X-Frame-Options", "DENY") w.Header().Set("Referrer-Policy", "no-referrer") next.ServeHTTP(w, r) }) } func env(k, def string) string { v := os.Getenv(k) if v == "" { return def } return v } func envBool(k string, def bool) bool { v := strings.TrimSpace(strings.ToLower(os.Getenv(k))) if v == "" { return def } switch v { case "1", "true", "yes", "y", "on": return true case "0", "false", "no", "n", "off": return false default: return def } } func envDur(k string, def time.Duration) time.Duration { v := os.Getenv(k) if v == "" { return def } d, err := time.ParseDuration(v) if err != nil { return def } return d } func envInt(k string, def int) int { v := os.Getenv(k) if v == "" { return def } n, err := strconv.Atoi(strings.TrimSpace(v)) if err != nil { return def } return n }