mesh-host/cmd/mesh-host/seatlist.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync/atomic"
	"time"

	conditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
	condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/seatdef"
	"git.novox.be/novox/mesh-sdk/go/seatlist"
	"git.novox.be/novox/mesh-sdk/go/views"
	"github.com/novox/mesh-host/internal/link"
	"github.com/novox/mesh-host/internal/seatwait"
)

// The node-engine reads the seat list as every part does (novox/hq ADR 0635 §1, ADR 0636): through the seat mesh-catalog,
// from the bootstrapped definitions until it has been read, acting on nothing new meanwhile (the composer's gate), and
// following the seat's events. What it holds is given to catalogue at every change, which compiles no seat list.

// seatListNow is this run's reader, what the node-engine's verbs seat-list and seat-list-again answer from.
var seatListNow atomic.Pointer[seatListState]

// seatListRead answers the verb seat-list: what the engine holds of the list and what it waits for (ADR 0636 §7).
func seatListRead() (any, error) {
	st := seatListNow.Load()
	if st == nil {
		return nil, errors.New("this engine's node agent has not started, so it holds no seat list yet")
	}
	status := st.reader.Status()
	return map[string]any{"status": status, "health": status.Health()}, nil
}

// seatListAgain answers the verb seat-list-again: the list read now from the seat, and what is held afterwards.
func seatListAgain() (any, error) {
	st := seatListNow.Load()
	if st == nil {
		return nil, errors.New("this engine's node agent has not started, so it holds no seat list yet")
	}
	err := st.reader.Reload()
	st.waiter.Changed() // a read that changed nothing held still ends the wait
	out, _ := seatListRead()
	if err != nil {
		out.(map[string]any)["error"] = err.Error()
	}
	return out, nil
}

// seatListState is one run's reader, the finding that it waits, and the channel on which the composer hears a change.
type seatListState struct {
	reader  *seatlist.Reader
	waiter  *seatwait.Waiter
	changed chan struct{}
}

// sessionBus is the reader's bus over the link: a verb of the seat is asked in the engine's own name, an event heard on
// the seat's event subject for as long as ctx lives.
type sessionBus struct {
	ctx     context.Context
	session *link.Session
}

func (b sessionBus) Ask(key string, body any) (json.RawMessage, error) {
	seat, verb, _ := strings.Cut(strings.TrimPrefix(key, "seat:"), ".")
	return b.session.AskMeshSeat(b.ctx, seat, verb, body)
}

func (b sessionBus) Subscribe(pattern string, handle func(string, json.RawMessage) error) error {
	seat, event, _ := strings.Cut(pattern, ".")
	heard, stop, err := b.session.Hear(seatdef.EventSubject(seat, event))
	if err != nil {
		return err
	}
	go func() {
		defer stop()
		for {
			select {
			case <-b.ctx.Done():
				return
			case body, ok := <-heard:
				if !ok {
					return
				}
				_ = handle(pattern, body)
			}
		}
	}()
	return nil
}

// seatWaitClock is the clock of the node's bound; the tests give their own.
var seatWaitClock seatwait.Clock = realClock{}

type realClock struct{}

func (realClock) Now() time.Time { return time.Now() }
func (realClock) AfterFunc(d time.Duration, f func()) seatwait.Timer {
	return time.AfterFunc(d, f)
}

// findingsOfTheNode writes the node's findings in its state `findings`, in the layout of the seat mesh-conditions.
type findingsOfTheNode struct{ session *link.Session }

func (w findingsOfTheNode) encode(f views.Finding) (string, []byte, error) {
	key := conditions.Key(conditions.ScopeMachine, f.ID, f.Kind)
	v := condseat.FindingsByKeyValue{V: 1, Key: key, Kind: f.Kind, Observer: f.Observer, Severity: conditions.Warning,
		Subject: condseat.Subject{Scope: conditions.ScopeMachine, ID: f.ID, Machine: f.ID}, Summary: f.Summary,
		Headline: f.Headline, Explanation: f.Explanation, Source: "seat-list-unread", At: f.At.UTC()}
	if err := conditions.CheckFinding(v); err != nil {
		return "", nil, err
	}
	raw, err := json.Marshal(v)
	return key, raw, err
}

func (w findingsOfTheNode) Put(f views.Finding) error {
	key, raw, err := w.encode(f)
	if err != nil {
		return err
	}
	return w.session.PutFinding(key, raw)
}

func (w findingsOfTheNode) Delete(f views.Finding) error {
	key, _, err := w.encode(f)
	if err != nil {
		return err
	}
	return w.session.DeleteFinding(key)
}

// startSeatList starts this run's reader: its bootstrapped seats are given to catalogue at once, and every change of what
// it holds again, saying so to the composer and to the waiter that writes the five-minute finding, and stating again what
// the engine reads (link.Session.RestateReads, ADR 0618 §4): the seats it states are those of the list it holds.
func startSeatList(ctx context.Context, session *link.Session, node string, say func(string)) (*seatListState, error) {
	st := &seatListState{changed: make(chan struct{}, 1)}
	// Start returns at once and reads the seat on a goroutine, which may announce a change before the waiter exists.
	var announce atomic.Pointer[func()]
	reader, err := seatlist.Start(seatlist.Config{Bus: sessionBus{ctx, session}, Bootstrap: seatlist.Bootstrapped(),
		OnChange: func(seats []catalogue.Seat) {
			catalogue.UseSeats(seats)
			// What the engine reads changed with the list: its record and its event say it (ADR 0618 §4), off the
			// reader's goroutine, which returns quickly.
			go session.RestateReads(ctx)
			if f := announce.Load(); f != nil { // the first read may end before the waiter exists
				(*f)()
			}
		}})
	if err != nil {
		return nil, fmt.Errorf("the node-engine cannot follow the seat list: %w", err)
	}
	st.reader = reader
	st.waiter = &seatwait.Waiter{Node: node, Observer: "node-engine", Reader: reader, Findings: findingsOfTheNode{session},
		Clock: seatWaitClock, Say: say}
	told := func() {
		st.waiter.Changed()
		select {
		case st.changed <- struct{}{}:
		default:
		}
	}
	announce.Store(&told)
	catalogue.UseSeats(reader.Seats())
	session.RestateReads(ctx) // its first statement holds the bootstrapped seats at least (issue 1966)
	told()
	// The first read ends whether the seat answered or not, and a read that held what the reader held already says no
	// change: its end is told to the waiter and the composer, so a status that moved from reading to heard starts or ends
	// the bound (ADR 0636 §1). Start's return held nobody's start.
	go func() {
		select {
		case <-reader.Ready():
			catalogue.UseSeats(reader.Seats())
			session.RestateReads(ctx)
			told()
		case <-ctx.Done():
		}
	}()
	seatListNow.Store(st)
	return st, nil
}