mesh-host/internal/link/nodehealth.go

package link

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"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 layout moves into mesh-sdk before the push (the lead, 2026-10-29); until then it is this package's.

// HealthBucket is the node-engine's declared state health: key `<node>`, value NodeHealth.
const HealthBucket = "mesh-host_health"

// NodeHealthVersion is the version of NodeHealth's layout.
const NodeHealthVersion = 1

// NodeHealth is one node's health report, as its engine last found it.
type NodeHealth struct {
	V    int    `json:"v"`
	Node string `json:"node"`
	// Network is the machine's networking, its parts as Health.Network says them, without the streaks; nil where this
	// engine does not judge it yet.
	Network *NetworkHealth `json:"network,omitempty"`
	// Peers are the mesh tunnel's peers on the last look that read them, sorted by key.
	Peers []PeerSeen `json:"peers,omitempty"`
	// Units is the machine's service managers: each failed unit no module places (Health.Units).
	Units *UnitsHealth `json:"units,omitempty"`
	// Relogin are the accounts whose running session needs a new login for a group it was given.
	Relogin []NodeAccount `json:"relogin,omitempty"`
	// Found are the directories used as found.
	Found []NodeDirectory `json:"found,omitempty"`
	// At is when the engine wrote this, on a change: never a reason to write.
	At time.Time `json:"at"`
}

// PeerSeen is one peer of the mesh's tunnel: its public key, its allowed addresses, and whether this machine saw a
// handshake with it within five minutes.
type PeerSeen struct {
	Key       string   `json:"key"`
	Addresses []string `json:"addresses,omitempty"`
	Seen      bool     `json:"seen"`
}

// NodeAccount is one account marked relogin needed, with the module whose resource it is and why.
type NodeAccount struct {
	Account  string    `json:"account"`
	Module   string    `json:"module,omitempty"`
	Resource string    `json:"resource,omitempty"`
	Reason   string    `json:"reason"`
	Since    time.Time `json:"since"`
}

// NodeDirectory is one directory used as found: the module declaring it, its resource and path, and why.
type NodeDirectory struct {
	Module   string    `json:"module"`
	Resource string    `json:"resource"`
	Path     string    `json:"path"`
	Reason   string    `json:"reason"`
	Since    time.Time `json:"since"`
}

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

// NodeHealthOf is the report a health statement and the tunnel's peers make: the statement's network without its
// streaks, its machine units, and each account and directory of its resources that the six alerts read.
func NodeHealthOf(node string, h Health, peers []PeerSeen) 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
			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
}

// EncodeNodeHealth is h as JSON at its version.
func EncodeNodeHealth(h NodeHealth) ([]byte, error) {
	h.V = NodeHealthVersion
	if h.Node == "" {
		return nil, errors.New("a node's health report names its node")
	}
	return json.Marshal(h)
}

// DecodeNodeHealth reads a report.
func DecodeNodeHealth(raw []byte) (NodeHealth, error) {
	var h NodeHealth
	if err := json.Unmarshal(raw, &h); err != nil {
		return NodeHealth{}, fmt.Errorf("a node's health report is not readable: %w", err)
	}
	if h.V != NodeHealthVersion {
		return NodeHealth{}, fmt.Errorf("a node's health report of version %d, and this reads %d", h.V, NodeHealthVersion)
	}
	return h, nil
}

// SameNodeHealth says two reports say the same, whenever each was written.
func SameNodeHealth(a, b NodeHealth) bool {
	a.At, b.At = time.Time{}, time.Time{}
	ra, errA := json.Marshal(a)
	rb, errB := json.Marshal(b)
	return errA == nil && errB == nil && bytes.Equal(ra, rb)
}

// 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 := DecodeNodeHealth(raw); err == nil {
				*last = &was
			}
		}
	}
	if *last != nil && SameNodeHealth(**last, h) {
		return false, nil
	}
	h.At = now().UTC()
	raw, err := 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
}