mesh-host/internal/link/nodehealth.go

package link

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"time"

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go"
)

// The node-engine's own health report in the bus's own storage (issue 774, the operator's alert table, 2026-10-29): one
// key per node, `<node>`, of the engine's declared state `health` (HealthBucket, created by the grants part from the
// engine's manifest as its state `acts` is), written by this node's engine alone and only when what it says changed.
// The watchman (seat mesh-watchdogs) reads it for six alerts the controller raised from the passing health message:
//
//   - machine-network: Network, the machine's route, tunnel, name servers, resolver file and bus, each a part;
//   - network-rewritten: the resolver file's part, unhealthy with Writer, the program that rewrote it, and Owner, the
//     module whose file it is;
//   - network-unreachable: Peers, each peer of the mesh's tunnel and whether this machine saw it within five minutes,
//     read across every node's report;
//   - machine-units: Units, the failed system units no module owns;
//   - relogin-needed: Relogin, each account whose running session lacks a group it was given;
//   - directory-used-as-found: Found, each directory a module declares that the engine uses as found.
//

// The report's layout is mesh-sdk's (delivery.NodeHealth): the bucket, its version and its parts.
const (
	HealthBucket      = delivery.NodeHealthBucket
	NodeHealthVersion = delivery.NodeHealthVersion
)

type (
	NodeHealth    = delivery.NodeHealth
	PeerSeen      = delivery.PeerSeen
	NodeAccount   = delivery.NodeAccount
	NodeDirectory = delivery.NodeDirectory
)

// ReasonReloginNeeded starts an account's reason when its session needs a new login (accounts.ReasonRelogin).
const ReasonReloginNeeded = "relogin needed"

// Machines names the mesh's machines for a report's targets: each node by its address on the private network, and the
// hub's node. The zero value names none, and every part is then the machine's own.
type Machines struct {
	ByAddress map[string]string
	Hub       string
}

// targets is the machines a part points at, as the controller holds it (mesh-controller machine_network.go): the hub's
// node for TowardHub, each resolver's node by its address; none when it points at nothing, or at this machine or an
// address no other mesh machine has, as one of them makes the part the machine's own.
func (m Machines) targets(node string, toward []string) []string {
	if len(toward) == 0 {
		return nil
	}
	var out []string
	for _, t := range toward {
		x := m.ByAddress[t]
		if t == TowardHub {
			x = m.Hub
		}
		if x == "" || x == node {
			return nil
		}
		if !slices.Contains(out, x) {
			out = append(out, x)
		}
	}
	return out
}

// TowardHub is what a part failing toward the hub names in Toward (network.TowardHub).
const TowardHub = "hub"

// NodeHealthOf is the report a health statement and the tunnel's peers make: the statement's network without its
// streaks, each unhealthy part naming the machines it points at (machines), its machine units, and each account and
// directory of its resources that the six alerts read.
func NodeHealthOf(node string, h Health, peers []PeerSeen, machines Machines) NodeHealth {
	out := NodeHealth{V: NodeHealthVersion, Node: node, Units: h.Units}
	if len(peers) > 0 {
		out.Peers = append([]PeerSeen(nil), peers...)
		sort.Slice(out.Peers, func(i, k int) bool { return out.Peers[i].Key < out.Peers[k].Key })
	}
	if h.Network != nil {
		n := NetworkHealth{State: h.Network.State, Since: h.Network.Since, Parts: make([]NetworkPart, 0, len(h.Network.Parts))}
		for _, p := range h.Network.Parts {
			p.Streak = 0 // a look's count moves every look; the report says a part's state, not how often it was seen
			p.Targets = nil
			if p.State == StateUnhealthy {
				p.Targets = machines.targets(node, p.Toward)
			}
			n.Parts = append(n.Parts, p)
		}
		out.Network = &n
	}
	for _, r := range h.Resources {
		switch {
		case r.Kind == KindAccount && r.State == StateUnhealthy && strings.HasPrefix(r.Reason, ReasonReloginNeeded):
			out.Relogin = append(out.Relogin, NodeAccount{Account: r.Account, Module: r.Module, Resource: r.Resource,
				Reason: r.Reason, Since: r.Since})
		case r.Kind == KindDirectory && strings.HasPrefix(r.Reason, ReasonUsedAsFound):
			out.Found = append(out.Found, NodeDirectory{Module: r.Module, Resource: r.Resource, Path: r.Target,
				Reason: r.Reason, Since: r.Since})
		}
	}
	sort.Slice(out.Relogin, func(i, k int) bool {
		return out.Relogin[i].Account+out.Relogin[i].Module < out.Relogin[k].Account+out.Relogin[k].Module
	})
	sort.Slice(out.Found, func(i, k int) bool { return out.Found[i].Path < out.Found[k].Path })
	return out
}

// HealthStore is where a node's report is kept: its own key of HealthBucket.
type HealthStore interface {
	// Get is the key's value, nil when it has none.
	Get(ctx context.Context, key string) ([]byte, error)
	Put(ctx context.Context, key string, value []byte) error
}

// healthStore binds HealthBucket over the link open now: bound, never created.
func (l *natsLink) healthStore() (HealthStore, error) {
	kv, err := l.js.KeyValue(HealthBucket)
	if err != nil {
		return nil, fmt.Errorf("the node-engine's state %s cannot be bound (it is created from the node-engine's "+
			"manifest; written again once it is): %w", HealthBucket, err)
	}
	return natsHealth{kv}, nil
}

type natsHealth struct{ kv nats.KeyValue }

func (n natsHealth) Get(_ context.Context, key string) ([]byte, error) {
	e, err := n.kv.Get(key)
	if errors.Is(err, nats.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return e.Value(), nil
}

func (n natsHealth) Put(_ context.Context, key string, value []byte) error {
	_, err := n.kv.Put(key, value)
	return err
}

// WriteNodeHealth writes this node's report to its key when it says something other than the key holds, and says
// whether it wrote. The key is read before the first write of a run, so a restart that finds the same report writes
// nothing; a store that cannot be bound or written is an error, and the next look tries again.
func (q *Session) WriteNodeHealth(ctx context.Context, h NodeHealth) (bool, error) {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	l, ok := bus.(interface{ healthStore() (HealthStore, error) })
	if bus == nil || !ok {
		return false, ErrNoLink
	}
	store, err := l.healthStore()
	if err != nil {
		return false, err
	}
	return writeNodeHealth(ctx, store, &q.healthWritten, h, time.Now)
}

// writeNodeHealth is WriteNodeHealth over a store: last is the report last written or read, nil before the first.
func writeNodeHealth(ctx context.Context, store HealthStore, last **NodeHealth, h NodeHealth, now func() time.Time) (bool, error) {
	if *last == nil {
		raw, err := store.Get(ctx, h.Node)
		if err != nil {
			return false, fmt.Errorf("the health report of %s cannot be read: %w", h.Node, err)
		}
		if raw != nil {
			if was, err := delivery.DecodeNodeHealth(raw); err == nil {
				*last = &was
			}
		}
	}
	if *last != nil && delivery.SameNodeHealth(**last, h) {
		return false, nil
	}
	h.At = now().UTC()
	raw, err := delivery.EncodeNodeHealth(h)
	if err != nil {
		return false, err
	}
	if err := store.Put(ctx, h.Node, raw); err != nil {
		return false, fmt.Errorf("the health report of %s cannot be written: %w", h.Node, err)
	}
	*last = &h
	return true, nil
}