reflector-dashboard/svxdr/nodelist/nodes_flat.go

// svxdr — live-data daemon for Reflector Dashboard (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

// All-Nodes flat layer: one total, deduped node list across the whole crawled
// network (federation via externalState.sources). Each node is attributed to
// the REAL reflector it is connected to (source tag; this box's own host is
// shown as the reflector's real name, never an IP). Mesh dupes collapse to one
// owning ref. Instead of re-shipping the whole list, this holds state
// server-side and streams only per-node deltas (added/changed/removed) over SSE.

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

	"github.com/redis/go-redis/v9"
)

// Federation aggregate (reflector-network.com/api/nodes): callsign -> node with
// the REAL reflector name. Populated by the gap loop; used here to (a) cover the
// whole network, not just what our own crawl reaches, and (b) give real ref
// names instead of a bare IP. UPPER(callsign) keyed.
var aggNodes = struct {
	mu sync.RWMutex
	m  map[string]flatNode
}{m: map[string]flatNode{}}

func setAggNodes(m map[string]flatNode) {
	aggNodes.mu.Lock()
	aggNodes.m = m
	aggNodes.mu.Unlock()
}

type flatNode struct {
	Callsign string `json:"callsign"`
	Ref      string `json:"ref"`
	TG       string `json:"tg"`
}

var flatState = struct {
	mu    sync.RWMutex
	nodes map[string]flatNode // keyed by UPPER(callsign)
	ver   uint64
}{nodes: map[string]flatNode{}}

// --- SSE subscriber hub ----------------------------------------------------
type flatSub struct{ ch chan []byte }

var flatHub = struct {
	mu   sync.Mutex
	subs map[*flatSub]struct{}
}{subs: map[*flatSub]struct{}{}}

func flatSubscribe() *flatSub {
	s := &flatSub{ch: make(chan []byte, 8)}
	flatHub.mu.Lock()
	flatHub.subs[s] = struct{}{}
	flatHub.mu.Unlock()
	return s
}
func flatUnsubscribe(s *flatSub) {
	flatHub.mu.Lock()
	delete(flatHub.subs, s)
	flatHub.mu.Unlock()
}
func flatBroadcast(msg []byte) {
	flatHub.mu.Lock()
	for s := range flatHub.subs {
		select {
		case s.ch <- msg:
		default: // slow client — drop; it will re-sync on next full snapshot/reconnect
		}
	}
	flatHub.mu.Unlock()
}

// tgOf extracts the "tg" (with "*" prefix for restrictedTG) from a node's JSON.
func tgOf(raw json.RawMessage) string {
	var n struct {
		TG           json.Number `json:"tg"`
		RestrictedTG bool        `json:"restrictedTG"`
	}
	if json.Unmarshal(raw, &n) != nil {
		return ""
	}
	s := n.TG.String()
	if s == "" {
		return ""
	}
	if n.RestrictedTG {
		return "*" + s
	}
	return s
}

var localRefName string // real name of THIS reflector (from svxreflector.conf), never an IP

// buildFlat assembles the deduped map straight from the federation crawl. Each
// node is attributed to the REAL reflector it is connected to (source.Tag).
// First ref by tag wins (sources are tag-sorted → deterministic). A source
// whose tag is this box's own host is renamed to the reflector's real name so
// the ref column is never "127.0.0.1"/an IP. No local-status overlay — that mix
// carried trunk-side (remote) nodes and mis-tagged them as local.
func buildFlat() map[string]flatNode {
	out := make(map[string]flatNode, 256)
	selfHost := hostFromURL(reflectorURL)
	externalState.mu.RLock()
	for _, s := range externalState.sources {
		ref := s.Tag
		// Never expose a raw http(s)://host:port/status URL as the ref — those
		// are FEDERATION_SEEDS whose tag is the seed URL. Show only the host.
		if strings.HasPrefix(ref, "http://") || strings.HasPrefix(ref, "https://") {
			ref = hostFromURL(ref)
		}
		if (hostFromURL(s.URL) == selfHost || hostFromURL(s.Tag) == selfHost) && localRefName != "" {
			ref = localRefName
		}
		// Bare IP → prefer a hostname alias (a name resolving to the same IP).
		if net.ParseIP(ref) != nil {
			for _, a := range s.Aliases {
				if a != "" && net.ParseIP(a) == nil {
					ref = a
					break
				}
			}
		}
		for cs, raw := range s.Nodes {
			key := strings.ToUpper(cs)
			if _, seen := out[key]; seen {
				continue
			}
			out[key] = flatNode{Callsign: cs, Ref: ref, TG: tgOf(raw)}
		}
	}
	externalState.mu.RUnlock()

	// Cover the WHOLE network (not just what our own crawl reaches): fold in the
	// federation aggregate. Crawl wins for nodes we see live (fresh TG); the
	// aggregate adds everyone else AND supplies a real reflector name where the
	// crawl only had a bare IP.
	aggNodes.mu.RLock()
	for k, an := range aggNodes.m {
		if ex, ok := out[k]; ok {
			if an.Ref != "" && net.ParseIP(ex.Ref) != nil {
				ex.Ref = an.Ref
				out[k] = ex
			}
			continue
		}
		out[k] = flatNode{Callsign: an.Callsign, Ref: an.Ref}
	}
	aggNodes.mu.RUnlock()
	return out
}

// rebuildFlat rebuilds, diffs against the current state and — only when the set
// actually changed — bumps the version and streams the delta to SSE clients.
func rebuildFlat() {
	next := buildFlat()

	flatState.mu.Lock()
	prev := flatState.nodes
	var added, changed []flatNode
	var removed []string
	for k, nv := range next {
		if pv, ok := prev[k]; !ok {
			added = append(added, nv)
		} else if pv.Ref != nv.Ref || pv.TG != nv.TG {
			changed = append(changed, nv)
		}
	}
	for k, pv := range prev {
		if _, ok := next[k]; !ok {
			removed = append(removed, pv.Callsign)
		}
	}
	if len(added) == 0 && len(changed) == 0 && len(removed) == 0 {
		flatState.mu.Unlock()
		return
	}
	flatState.nodes = next
	flatState.ver++
	ver := flatState.ver
	flatState.mu.Unlock()

	payload, _ := json.Marshal(struct {
		Added   []flatNode `json:"added"`
		Changed []flatNode `json:"changed"`
		Removed []string   `json:"removed"`
		Ver     uint64     `json:"v"`
	}{added, changed, removed, ver})
	flatBroadcast([]byte("event: nodes-delta\ndata: " + string(payload) + "\n\n"))
}

// flatSnapshotJSON returns the full current list (for /api/nodes and the SSE
// initial frame), sorted by callsign.
func flatSnapshotJSON() ([]byte, uint64) {
	flatState.mu.RLock()
	rows := make([]flatNode, 0, len(flatState.nodes))
	for _, v := range flatState.nodes {
		rows = append(rows, v)
	}
	ver := flatState.ver
	flatState.mu.RUnlock()
	sort.Slice(rows, func(i, j int) bool { return rows[i].Callsign < rows[j].Callsign })
	b, _ := json.Marshal(struct {
		Nodes []flatNode `json:"nodes"`
		Ver   uint64     `json:"v"`
	}{rows, ver})
	return b, ver
}

// nodeFlatStart wires the endpoints and the update driver. Called from main()
// once rdb exists. The driver rebuilds on every pubsubChan tick (instant local
// changes) and on a slow safety ticker (catches federation crawl updates).
func nodeFlatStart(mux *http.ServeMux, rdb *redis.Client) {
	localRefName, _ = localReflectorMeta() // real reflector name for the "local" tag
	rebuildFlat()                          // prime

	mux.HandleFunc("/api/nodes", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json; charset=utf-8")
		b, _ := flatSnapshotJSON()
		writeMaybeGzip(w, r, b)
	})

	mux.HandleFunc("/api/nodes-stream", func(w http.ResponseWriter, r *http.Request) {
		flusher, ok := w.(http.Flusher)
		if !ok {
			http.Error(w, "SSE not supported", 500)
			return
		}
		w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
		w.Header().Set("Cache-Control", "no-cache")
		w.Header().Set("Connection", "keep-alive")

		// initial full snapshot
		snap, _ := flatSnapshotJSON()
		fmt.Fprintf(w, "event: nodes-full\ndata: %s\n\n", snap)
		flusher.Flush()

		sub := flatSubscribe()
		defer flatUnsubscribe(sub)
		ping := time.NewTicker(25 * time.Second)
		defer ping.Stop()

		for {
			select {
			case <-r.Context().Done():
				return
			case <-ping.C:
				fmt.Fprint(w, "event: ping\ndata: 1\n\n")
				flusher.Flush()
			case msg := <-sub.ch:
				w.Write(msg)
				flusher.Flush()
			}
		}
	})

	// Driver: instant on pubsub, plus a safety ticker for crawl-side changes.
	go func() {
		sub := rdb.Subscribe(ctx, pubsubChan)
		defer sub.Close()
		ch := sub.Channel()
		tick := time.NewTicker(3 * time.Second)
		defer tick.Stop()
		for {
			select {
			case <-ch:
				rebuildFlat()
			case <-tick.C:
				rebuildFlat()
			}
		}
	}()
}