bootnode/fleet/fleet.go

// Package fleet observes the managed luxd fleet read-only and renders it as
// the document the lux.cloud dashboard polls.
//
// One concern per file:
//
//	fleet.go    the wire document and how a network rolls up into it
//	luxd.go     talking to one luxd over its /v1 HTTP RPC
//	cluster.go  reading the Kubernetes facts (which pods, what the operator declared)
//
// The collector runs INSIDE the cluster that hosts the fleet, so a node's dial
// address is its pod IP — there is no endpoint file to go stale. Its public
// address is whatever the node itself advertises (info.getNodeIP), which is the
// only source that cannot drift from reality.
package fleet

import (
	"context"
	"fmt"
	"sort"
	"sync"
	"time"
)

// Networks is the managed set, in display order. Ports and IDs mirror the
// luxd StatefulSet in each namespace.
var Networks = []Network{
	{Name: "mainnet", Namespace: "lux-mainnet", HTTPPort: 9630, NetworkID: 1, ChainID: 96369},
	{Name: "testnet", Namespace: "lux-testnet", HTTPPort: 9640, NetworkID: 2, ChainID: 96368},
	{Name: "devnet", Namespace: "lux-devnet", HTTPPort: 9650, NetworkID: 3, ChainID: 96370},
}

// Network is one managed luxd network.
type Network struct {
	Name      string
	Namespace string
	HTTPPort  int
	NetworkID int
	ChainID   int
}

// TrackedChains are the luxd chains the dashboard shows, in display order.
var TrackedChains = []string{"C", "P", "X", "D"}

// ---------------------------------------------------------------------------
// Wire document. Field names and null-vs-absent are the contract with
// console/lib/fleet.ts — every field is always present, absent data is null.
// ---------------------------------------------------------------------------

// ChainHeight is one chain's bootstrap state and height on one node.
type ChainHeight struct {
	Chain        string `json:"chain"`
	Bootstrapped *bool  `json:"bootstrapped"`
	Height       *int64 `json:"height"`
}

// Node is one luxd validator pod.
type Node struct {
	PodName      string        `json:"pod_name"`
	PodIndex     int           `json:"pod_index"`
	Phase        string        `json:"phase"`
	Ready        bool          `json:"ready"`
	IP           *string       `json:"ip"`
	RPCReachable bool          `json:"rpc_reachable"`
	NodeID       *string       `json:"node_id"`
	Version      *string       `json:"version"`
	GitCommit    *string       `json:"git_commit"`
	DBVersion    *string       `json:"db_version"`
	Peers        *int          `json:"peers"`
	Healthy      *bool         `json:"healthy"`
	Chains       []ChainHeight `json:"chains"`
}

// Dex is the native D-Chain venue summary for a network.
type Dex struct {
	DChainBootstrapped *bool   `json:"d_chain_bootstrapped"`
	Venue              string  `json:"venue"`
	Markets            *int    `json:"markets"`
	Note               *string `json:"note"`
}

// NetworkFleet is one network's rollup.
//
// image is what the operator DECLARES (the LuxNetwork CR tag); version is what
// the nodes actually REPORT. They diverge whenever the CR is stale, and the
// dashboard must show the running truth, so observation wins for version.
type NetworkFleet struct {
	Network    string   `json:"network"`
	Namespace  string   `json:"namespace"`
	NetworkID  *int     `json:"network_id"`
	ChainID    *int     `json:"chain_id"`
	Phase      string   `json:"phase"`
	Image      *string  `json:"image"`
	Version    *string  `json:"version"`
	PluginSet  *string  `json:"plugin_set"`
	VMPlugins  []string `json:"vm_plugins"`
	Validators int      `json:"validators"`
	NodesTotal int      `json:"nodes_total"`
	NodesReady int      `json:"nodes_ready"`
	NodesRPCOK int      `json:"nodes_rpc_ok"`
	TotalPeers int      `json:"total_peers"`
	Dex        Dex      `json:"dex"`
	Nodes      []Node   `json:"nodes"`
	Error      *string  `json:"error"`
}

// Overview is the single read the dashboard polls.
type Overview struct {
	Cluster     string         `json:"cluster"`
	Networks    []NetworkFleet `json:"networks"`
	Totals      map[string]int `json:"totals"`
	GeneratedAt float64        `json:"generated_at"`
}

// ---------------------------------------------------------------------------
// Collector
// ---------------------------------------------------------------------------

// Collector assembles the overview from a Cluster and a Prober. Both are
// interfaces so the rollup can be tested without a cluster or a chain.
type Collector struct {
	Cluster Cluster
	Prober  Prober
	Markets Markets // may be nil; markets then stays null
}

// Overview collects every named network, or all of them when names is empty.
func (c *Collector) Overview(ctx context.Context, names []string) Overview {
	selected := selectNetworks(names)
	out := Overview{
		Cluster:     "lux-k8s",
		Networks:    make([]NetworkFleet, len(selected)),
		GeneratedAt: float64(time.Now().UnixNano()) / 1e9,
	}

	var wg sync.WaitGroup
	for i, n := range selected {
		wg.Add(1)
		go func() {
			defer wg.Done()
			out.Networks[i] = c.network(ctx, n)
		}()
	}
	wg.Wait()

	totals := map[string]int{"networks": len(out.Networks)}
	for _, f := range out.Networks {
		totals["validators"] += f.Validators
		totals["nodes_total"] += f.NodesTotal
		totals["nodes_ready"] += f.NodesReady
		totals["nodes_rpc_ok"] += f.NodesRPCOK
		totals["total_peers"] += f.TotalPeers
	}
	out.Totals = totals
	return out
}

func selectNetworks(names []string) []Network {
	if len(names) == 0 {
		return Networks
	}
	var out []Network
	for _, want := range names {
		for _, n := range Networks {
			if n.Name == want {
				out = append(out, n)
			}
		}
	}
	return out
}

func (c *Collector) network(ctx context.Context, n Network) NetworkFleet {
	f := NetworkFleet{
		Network:   n.Name,
		Namespace: n.Namespace,
		NetworkID: ptr(n.NetworkID),
		ChainID:   ptr(n.ChainID),
		Phase:     "Unknown",
		VMPlugins: []string{},
		Nodes:     []Node{},
		Dex:       Dex{Venue: "native-dchain"},
	}

	pods, podErr := c.Cluster.Pods(ctx, n.Namespace)
	if declared, err := c.Cluster.Declared(ctx, n.Namespace); err == nil {
		f.Phase = declared.Phase
		f.Validators = declared.Validators
		f.Image = strPtr(declared.Image)
		f.PluginSet = strPtr(declared.PluginSet)
		f.VMPlugins = declared.VMPlugins
	}

	sort.Slice(pods, func(i, j int) bool { return pods[i].Index < pods[j].Index })
	f.Nodes = make([]Node, len(pods))
	var wg sync.WaitGroup
	for i, p := range pods {
		f.Nodes[i] = Node{
			PodName: p.Name, PodIndex: p.Index, Phase: p.Phase, Ready: p.Ready,
			Chains: emptyChains(),
		}
		if p.IP == "" {
			continue
		}
		wg.Add(1)
		go func() {
			defer wg.Done()
			c.Prober.Probe(ctx, fmt.Sprintf("http://%s:%d", p.IP, n.HTTPPort), &f.Nodes[i])
		}()
	}
	wg.Wait()

	for i := range f.Nodes {
		if f.Nodes[i].Ready {
			f.NodesReady++
		}
		if f.Nodes[i].RPCReachable {
			f.NodesRPCOK++
		}
		if f.Nodes[i].Peers != nil {
			f.TotalPeers += *f.Nodes[i].Peers
		}
	}
	f.NodesTotal = len(f.Nodes)
	if f.Validators == 0 {
		f.Validators = f.NodesTotal
	}
	if v := observedVersion(f.Nodes); v != "" {
		f.Version = &v
	} else if f.Image != nil {
		f.Version = strPtr(imageTag(*f.Image))
	}
	if f.Phase == "Unknown" && f.NodesRPCOK > 0 {
		f.Phase = "Running"
	}
	f.Dex = c.dex(ctx, n.Name, f.Nodes)
	switch {
	case podErr != nil:
		f.Error = strPtr("cluster read failed: " + podErr.Error())
	case f.NodesTotal == 0:
		f.Error = strPtr("no luxd pods found in " + n.Namespace)
	}
	return f
}

// dex rolls the D-Chain bootstrap state up across nodes: true if any node has
// it, false if every reporting node says no, null if nothing reported.
func (c *Collector) dex(ctx context.Context, network string, nodes []Node) Dex {
	d := Dex{Venue: "native-dchain"}
	for _, n := range nodes {
		for _, ch := range n.Chains {
			if ch.Chain != "D" || ch.Bootstrapped == nil {
				continue
			}
			if d.DChainBootstrapped == nil || !*d.DChainBootstrapped {
				d.DChainBootstrapped = ch.Bootstrapped
			}
		}
	}
	if c.Markets != nil {
		if m, ok := c.Markets.Count(ctx, network); ok {
			d.Markets = &m
		}
	}
	if d.DChainBootstrapped != nil && !*d.DChainBootstrapped {
		d.Note = strPtr("D-Chain not yet bootstrapped")
	}
	return d
}

func emptyChains() []ChainHeight {
	out := make([]ChainHeight, len(TrackedChains))
	for i, c := range TrackedChains {
		out[i] = ChainHeight{Chain: c}
	}
	return out
}

// observedVersion is the version the running fleet reports — the lowest-index
// node that answers wins, so the value is stable across polls.
func observedVersion(nodes []Node) string {
	for _, n := range nodes {
		if n.Version != nil && *n.Version != "" {
			return *n.Version
		}
	}
	return ""
}

func imageTag(image string) string {
	for i := len(image) - 1; i >= 0; i-- {
		if image[i] == ':' {
			return image[i+1:]
		}
		if image[i] == '/' {
			break
		}
	}
	return image
}

func ptr[T any](v T) *T { return &v }

func strPtr(s string) *string {
	if s == "" {
		return nil
	}
	return &s
}