mesh-host/internal/link/session.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	sdkkeys "git.novox.be/novox/mesh-sdk/go/keys"
	"strings"
	"sync"
	"time"

	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/health"
)

// Session is the node-engine's hold on the bus (novox/hq ADR 0539 §9): the link open now, what is asked of this
// machine over it, and what the machine says on it unasked: its health and its haves.
//
// **It applies nothing and orders nothing.** A module is applied by the node agent from its want. Nothing here
// receives, orders or reconciles a whole declaration.
type Session struct {
	// Fallback is the engine's fall back to its old login its health report carries now (Rotator.Fallback); nil says none.
	Fallback func() *nodeengine.LoginFallback
	// linkedUser is the bus user the link open now authenticated as, guarded by mu (WhenLinkedAs).
	linkedUser string
	// 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
	// Acts is the node seat node-engine this engine holds (novox/hq ADR 0563): served on every link, and its asks to the
	// operator asked and heard over it. Nil serves no seat.
	Acts *HandActs
	// Keys reads the node's public keys and its certificate for its record (ADR 0555 §3), at each facts report; nil writes
	// no record.
	Keys    func() RecordKeys
	Say     Announce
	Timeout time.Duration
	// RefusedPause is how long a link keeps running after the bus refused it a subscription before it is dropped and
	// opened again (zero is DefaultRefusedPause); doubled for each link in a row that is refused again, to RefusedPauseMost.
	RefusedPause time.Duration

	// refusedInARow is how many links in a row the bus refused a subscription to.
	refusedInARow int

	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

	// runnerStated is the capabilities this machine's tool runner last stated of the calls it answers (askRunner), said in the
	// node's record. Held under saying.
	runnerStated []string

	// saying is held while a facts event is noted, so the haves are written in the order the facts are made.
	saying sync.Mutex

	// reads says whether the engine has said its manifest reads on a link before (novox/hq ADR 0618 §4).
	reads struct {
		sync.Mutex
		linked bool
	}
	// restating takes RestateReads one at a time.
	restating sync.Mutex

	// haves is what this node-engine writes beside its facts (novox/hq ADR 0429 §4, issue 1132).
	haves haveState
	// healthWritten is this node's health report as last written or read (WriteNodeHealth): the health loop alone
	// writes it, one look at a time.
	healthWritten *NodeHealth
}

// DefaultRefusedPause and RefusedPauseMost bound how soon a link refused a subscription is opened again: the grants
// part composes the user list again within a minute of what it is composed from changing, so the first pause waits for
// that; one that is still refused waits longer each time, never more than half an hour, and always says so.
const (
	DefaultRefusedPause = time.Minute
	RefusedPauseMost    = 30 * time.Minute
)

// refusedPause is the pause before a link the bus refused a subscription to is opened again, counting it.
func (q *Session) refusedPause() time.Duration {
	q.mu.Lock()
	defer q.mu.Unlock()
	pause := q.RefusedPause
	if pause <= 0 {
		pause = DefaultRefusedPause
	}
	for i := 0; i < q.refusedInARow && pause < RefusedPauseMost; i++ {
		pause *= 2
	}
	q.refusedInARow++
	return min(pause, RefusedPauseMost)
}

// refusalsHeld says a link has held its subscriptions: the next refusal starts from the first pause again.
func (q *Session) refusalsHeld() {
	q.mu.Lock()
	q.refusedInARow = 0
	q.mu.Unlock()
}

func (q *Session) init() {
	q.once.Do(func() {
		if q.Say == nil {
			q.Say = func(string) {}
		}
		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 in its haves on the link open now, in the order they are made, and whether a link
// was open to say them on. A facts event made with no link open is not kept: the next link says the facts as they are
// then. They are said in the haves alone: the CONTROL stream they were also published on was the controller's, which
// ended by its unassignment (issue 1126). A write the bucket refuses is said and written again (issue 1132).
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
	}
	r.Node = q.Membership.Node
	q.askRunner(ctx, bus)
	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.linkedUser = ""
	if u, ok := bus.(interface{ User() string }); ok {
		q.linkedUser = u.User()
	}
	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
		q.linkedUser = ""
		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)
}

// 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
}

// PublishAsk publishes an ask or a cancel to the operator's router over the link open now (novox/hq ADR 0563 §3).
func (q *Session) PublishAsk(subject string, body any, id string) error {
	asking, err := q.operatorAsks()
	if err != nil {
		return err
	}
	return asking.PublishAsk(subject, body, id)
}

// RouterRecord reads one key of the router's records over the link open now.
func (q *Session) RouterRecord(key string) (*stdio.StateEntry, error) {
	asking, err := q.operatorAsks()
	if err != nil {
		return nil, err
	}
	return asking.RouterRecord(key)
}

// StatedReads is what this engine states it reads of a manifest and when it made that statement, at its start or at the
// last change of its seat list that changed it (the node seat's verb manifest-reads, novox/hq ADR 0618 §3): the record is
// written only when it differs, so its put says nothing of when an unchanged statement was made.
func (q *Session) StatedReads() ([]string, time.Time) { return stated.now() }

// RestateReads states what this engine reads again, after the seat list it holds changed (novox/hq ADR 0618 §4, ADR
// 0636): when the statement changed, its node record carries the new one at once, written where it differs, and the node
// seat's event manifest-reads-changed says it ("started": what it states changed), so the assigner and the builder ask
// the verb again. A statement that did not change writes and says nothing. Calls are taken one at a time, so the last
// write is the newest statement.
func (q *Session) RestateReads(ctx context.Context) {
	q.init()
	q.restating.Lock()
	defer q.restating.Unlock()
	reads, changed := stated.restate()
	if !changed {
		return
	}
	s := &q.haves
	s.mu.Lock()
	if s.node != nil && s.node.Record != nil {
		node, rec := *s.node, *s.node.Record
		rec.ManifestReads = reads
		node.Record = &rec
		s.node = &node
	}
	s.mu.Unlock()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	if bus != nil {
		q.writeHaves(ctx, bus)
	}
	q.SayEngine(nodeengine.EventManifestReadsChanged, nodeengine.ManifestReadsChangedEvent{V: sdkkeys.Version,
		Node: q.Membership.Node, ManifestReads: reads, At: time.Now().UTC().Format(time.RFC3339Nano), Why: ReadsStarted})
}

// The reasons the engine says its reads, in the event's why (novox/hq ADR 0618 §4): it started, or its node returned
// after being away (a link after the first).
const (
	ReadsStarted  = "started"
	ReadsReturned = "returned"
)

// SayReads says the node seat's event manifest-reads-changed once for the link just opened (novox/hq ADR 0618 §4):
// why "started" on the engine's first link and "returned" on every link after it, so a reader asks the verb again once
// per connection and never once per pass.
func (q *Session) SayReads() {
	q.reads.Lock()
	why := ReadsStarted
	if q.reads.linked {
		why = ReadsReturned
	}
	q.reads.linked = true
	q.reads.Unlock()
	q.SayEngine(nodeengine.EventManifestReadsChanged, nodeengine.ManifestReadsChangedEvent{V: sdkkeys.Version,
		Node: q.Membership.Node, ManifestReads: manifestReads(), At: time.Now().UTC().Format(time.RFC3339Nano), Why: why})
}

// SayEngine says one of the node seat's events over the link open now; with none open it is not said.
func (q *Session) SayEngine(event string, body any) {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	if sayer, ok := bus.(interface{ SayEngine(string, any) error }); ok {
		if err := sayer.SayEngine(event, body); err != nil {
			q.Say("could not say the node seat's " + event + ": " + err.Error())
		}
	}
}

func (q *Session) operatorAsks() (OperatorAsks, 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)
	}
	asking, ok := bus.(OperatorAsks)
	if !ok {
		return nil, errNoAsking
	}
	return asking, nil
}

// runnerAsk bounds the question to the tool runner.
const runnerAsk = 3 * time.Second

// askRunner reads what this machine's tool runner states of the calls it answers, asked of its own instance with the PING
// the witness asks (ENG-2, issue 1966), at each report: a runner replaced by a build that states more, or less, is said in
// the record from the next report. A runner that does not answer states what it stated before, since its absence says
// nothing of its calls; one that answers without stating any states none, as an older runner does. Called under saying.
func (q *Session) askRunner(ctx context.Context, bus Bus) {
	stater, ok := bus.(Stater)
	if !ok {
		return
	}
	asking, cancel := context.WithTimeout(ctx, runnerAsk)
	defer cancel()
	md, err := stater.Stated(asking, ToolRunnerService, q.Membership.Node)
	if err != nil {
		return
	}
	var stated []string
	for _, c := range strings.Split(md[CapabilitiesKey], ",") {
		if c = strings.TrimSpace(c); c != "" {
			stated = append(stated, c)
		}
	}
	q.runnerStated = stated
}