mesh-host/internal/link/session.go

package link

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

	"git.novox.be/novox/mesh-sdk/go/health"
)

// Writer is what the node-engine calls itself on the facts it says (novox/hq ADR 0227 §2, ADR 0539 §9): the program
// that wrote them, beside its epoch and sequence.
const Writer = "node-engine"

// Session is the node-engine's hold on the bus (novox/hq ADR 0539 §9): the link open now, what the controller asks
// of this machine over it, and what the machine says on it unasked: its health, its haves and its facts.
//
// **It applies nothing and orders nothing.** A module is applied by the node agent from its want, and the facts are an
// event of the node, ordered by the node-engine's own epoch and sequence. Nothing here receives, orders or reconciles
// a whole declaration.
type Session struct {
	// Membership is the node and the mesh's key a controller's ask is verified with.
	Membership Membership
	// OnLink is run, off the link's own loop, each time a link opens and when the controller asks the node to say its
	// facts again. It says them with SayFacts. Nil says nothing.
	OnLink func(ctx context.Context)
	// Heard is told every facts event the mesh has taken. Nil is allowed.
	Heard func(Report)
	// HaveWritten is told each time a have key is written, or found already saying what this node-engine would write:
	// the first proves a node-engine on trial (novox/hq ADR 0539 §9). Nil is allowed.
	HaveWritten func()
	// HandOver judges and records a directory used as found handed to the mesh by the operator, asked by
	// the controller over the link (novox/hq issue 356). Nil answers every such ask with a refusal.
	HandOver HandOverFunc
	// SetuidSearch throws away the last search for setuid programs and starts a full one, asked by the
	// controller's terminal over the link (novox/hq issue 361). Nil answers every such ask with a refusal.
	SetuidSearch SetuidSearchFunc
	// JudgeDirectories judges this node's declared directories read-only, asked by the controller over the
	// link (novox/hq issue 358). Nil answers every such ask with a refusal.
	JudgeDirectories JudgeDirectoriesFunc
	// AccessAct judges and runs the take act on an access's folder, asked by the controller on an approved
	// warrant (novox/hq ADR 0328 §7). Nil answers every such ask with a refusal.
	AccessAct AccessActFunc
	// MoveSecret moves one value sealed to this machine into the vault, asked by the controller during the move of
	// the secrets (design 57's V3, issue 676). Nil answers every such ask with a refusal.
	MoveSecret MoveSecretFunc
	Say        Announce
	Timeout    time.Duration

	once sync.Once

	mu  sync.Mutex
	bus Bus // the link open now; nil while there is none
	// gone is closed when the link open now closes, and changed whenever a link opens or closes (WhenLinked).
	gone    chan struct{}
	changed chan struct{}
	// owed is HealthOwed's: a module left motion unjudged.
	owed chan struct{}

	// haveAgain guards the one write of the haves pending after a transient failure (haveFailed), and how many came in a
	// row.
	haveAgain        sync.Mutex
	haveAgainPending bool
	haveAgainTries   int

	// saying is held while a facts event is made and said, so they leave in the order they are made.
	saying sync.Mutex
	// epoch is this process's own: the time it started, in milliseconds, so a later process orders after an earlier one
	// without anything kept on disk; sequence is one higher for every facts event it says (ADR 0227 §2).
	epoch    int64
	sequence int64

	// haves is what this node-engine writes beside its facts (novox/hq ADR 0429 §4, issue 1132).
	haves haveState
}

func (q *Session) init() {
	q.once.Do(func() {
		if q.Say == nil {
			q.Say = func(string) {}
		}
		q.epoch = time.Now().UnixMilli()
		q.changed = make(chan struct{})
		q.owed = make(chan struct{}, 1)
	})
}

// HealthOwed is told, at most once until it is read, when a module's change ended and no look has judged it since: the
// health loop looks and says its statement at once rather than at its next tick or five minutes on, so a module with
// nothing long-running is ok within a look of its apply, not five minutes after it (2026-10-25, review of fix 15).
func (q *Session) HealthOwed() <-chan struct{} {
	q.init()
	return q.owed
}

// oweHealth nudges the health loop; a nudge already waiting is enough.
func (q *Session) oweHealth() {
	q.init()
	select {
	case q.owed <- struct{}{}:
	default:
	}
}

// SayFacts says this node's facts on the link open now, in the node-engine's own order, and whether the bus took them.
// A facts event made with no link open is not kept: the next link says the facts as they are then.
func (q *Session) SayFacts(ctx context.Context, r Report) bool {
	q.init()
	q.saying.Lock()
	defer q.saying.Unlock()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	if bus == nil {
		return false
	}
	q.sequence++
	r.Node = q.Membership.Node
	r.Writer = Writer
	r.Order = Order{Epoch: q.epoch, Sequence: q.sequence}
	if !publishReport(context.WithoutCancel(ctx), bus, q.Membership, r, q.Say, q.Timeout) {
		return false
	}
	// After the facts, which are sent first and unchanged; a write the bucket refuses never undoes them (issue 1132).
	if q.noteApply(r) {
		q.writeHaves(ctx, bus)
	}
	if q.Heard != nil && r.Refused == "" {
		q.Heard(r)
	}
	return true
}

// FactsAsked is the mesh asking this node to say its facts again: run on the link open now.
func (q *Session) FactsAsked(ctx context.Context) {
	q.init()
	if q.OnLink != nil {
		go q.OnLink(ctx)
	}
}

// Attach is a link opening: it is the link facts are said on from now.
func (q *Session) Attach(ctx context.Context, bus Bus) {
	q.init()
	q.mu.Lock()
	q.bus = bus
	q.gone = make(chan struct{})
	close(q.changed)
	q.changed = make(chan struct{})
	q.mu.Unlock()
	q.FactsAsked(ctx)
}

// WithLink is ctx carrying the link now open, so an apply can fetch a composition's files given by digest (novox/hq ADR
// 0486, issue 1695). A session with no link open carries none.
func (q *Session) WithLink(ctx context.Context) context.Context {
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	if bus == nil {
		return ctx
	}
	return WithFiles(ctx, bus)
}

// Detach is the link closing: nothing more is said on it.
func (q *Session) Detach(bus Bus) {
	q.init()
	q.saying.Lock()
	defer q.saying.Unlock()
	q.mu.Lock()
	if q.bus == bus {
		q.bus = nil
		close(q.gone)
		close(q.changed)
		q.changed = make(chan struct{})
	}
	q.mu.Unlock()
}

// Linked says whether a link to the bus is open now (novox/hq ADR 0241: the bus is one part of the
// machine's networking its engine judges).
func (q *Session) Linked() bool {
	q.init()
	q.mu.Lock()
	defer q.mu.Unlock()
	return q.bus != nil
}

// SayHealth says a health statement on the link open now (novox/hq ADR 0240, to-be 48 §4), and whether
// the bus took it. A statement is a word about the machine as it is, and it must not wait behind an apply: a
// container crash-looping while a long apply runs is said while it runs. False with no link open, or one that cannot
// say it; the caller says it again on its next look, and the next facts event carries it whatever happens.
func (q *Session) SayHealth(ctx context.Context, h Health) bool {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	sayer, ok := bus.(HealthBus)
	if bus == nil || !ok {
		return false
	}
	body, err := json.Marshal(HealthSaid{Node: q.Membership.Node, Health: h})
	if err != nil {
		return false
	}
	saying, cancel := context.WithTimeout(ctx, q.timeout())
	defer cancel()
	if err := sayer.Health(saying, q.Membership.Node, body); err != nil {
		q.Say("could not say this machine's health: " + err.Error())
		return false
	}
	// A look that changed a module's health changes its have (issue 1132).
	if q.noteHealth(h) {
		q.writeHaves(ctx, bus)
	}
	return true
}

func (q *Session) timeout() time.Duration {
	if q.Timeout <= 0 {
		return 10 * time.Second
	}
	return q.Timeout
}

// AskTool asks a module's tool on this machine's own node tools, as a declared health check (novox/hq ADR
// 0240, to-be 48 §3), and reads its answer: healthy or not, with why. An error when there is no link open
// or nothing answered — said as the check's failure, never as healthy.
func (q *Session) AskTool(ctx context.Context, module, tool string) (health.Answer, error) {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	if bus == nil {
		return health.Answer{}, errors.New("no link to the bus is open")
	}
	asker, ok := bus.(ToolBus)
	if !ok {
		return health.Answer{}, errors.New("the link to the bus open now cannot ask a tool")
	}
	reply, err := asker.Ask(ctx, ToolSubject(module, tool, q.Membership.Node), []byte("{}"))
	if err != nil {
		return health.Answer{}, err
	}
	return ReadToolAnswer(reply)
}

// AskController asks the controller a line mesh-cli was given on this machine, on this node's own subject, over
// the link open now (novox/hq ADR 0272 §3), and returns its answer as it came. No link open is said, not waited for.
func (q *Session) AskController(ctx context.Context, body []byte) ([]byte, error) {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	if bus == nil {
		return nil, fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked)
	}
	asker, ok := bus.(ToolBus)
	if !ok {
		return nil, fmt.Errorf("%w: the link to the bus open now cannot ask the controller", ErrNotAsked)
	}
	return asker.Ask(ctx, CLISubject(q.Membership.Node), body)
}

// ErrNotAsked is a request the link knows it never delivered: no link open, or nobody listening on its subject. Any
// other failure of a request may come after it was delivered, so what came of it is not known (novox/hq ADR 0272).
var ErrNotAsked = errors.New("not asked")

// ReadToolAnswer reads a tool's reply envelope — `{result, error}`, as the node tools answer every call —
// as a health answer, through the SDK's reader (novox/hq ADR 0283): healthy, unhealthy, or waits for the
// operator with what it waits for. An error, and a result that is not an answer, are unhealthy; a wait the
// SDK cannot check is unhealthy too, so a malformed one never excuses anything.
func ReadToolAnswer(reply []byte) (health.Answer, error) {
	var env struct {
		Result json.RawMessage `json:"result"`
		Error  string          `json:"error"`
	}
	if err := json.Unmarshal(reply, &env); err != nil {
		return health.Answer{}, fmt.Errorf("its answer is not one: %w", err)
	}
	if env.Error != "" {
		return health.Unhealthy(env.Error), nil
	}
	a, err := health.Read(env.Result)
	if err != nil {
		return health.Unhealthy("it answered something other than healthy or not"), nil
	}
	return a, nil
}

// SeatToolSubject is where a node seat's verb is asked of one machine's holder (design 33 §4): the seat's verb subject
// with the machine as its last token.
func SeatToolSubject(seat, verb, node string) string {
	return "mesh.seat." + seat + ".tool." + verb + "." + node
}

// AskSeat asks the holder of a seat on this node a verb, over the link open now, and answers its result, or its refusal
// as an error. The one call the node agent makes of another part: the backup before a data module moves (ADR 0429 §14).
func (q *Session) AskSeat(ctx context.Context, seat, verb string, args any, within time.Duration) (json.RawMessage, error) {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	if bus == nil {
		return nil, fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked)
	}
	asker, ok := bus.(ToolBus)
	if !ok {
		return nil, fmt.Errorf("%w: the link to the bus open now cannot ask a seat", ErrNotAsked)
	}
	body, err := json.Marshal(args)
	if err != nil {
		return nil, err
	}
	asking, cancel := context.WithTimeout(ctx, within)
	defer cancel()
	reply, err := asker.Ask(asking, SeatToolSubject(seat, verb, q.Membership.Node), body)
	if err != nil {
		return nil, fmt.Errorf("%s.%s could not be asked: %w", seat, verb, err)
	}
	var env struct {
		Result json.RawMessage `json:"result"`
		Error  string          `json:"error"`
	}
	if err := json.Unmarshal(reply, &env); err != nil {
		return nil, fmt.Errorf("%s.%s answered with something that is not a tool's answer", seat, verb)
	}
	if env.Error != "" {
		return nil, errors.New(env.Error)
	}
	return env.Result, nil
}