mesh-host/internal/link/engineseat.go

package link

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

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/asks"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/micro"
)

// The node seat node-engine, served on the engine's own link (novox/hq ADR 0563 §1).
//
// Each verb is answered on this node's subject alone, `mesh.seat.node-engine.tool.<verb>.<node>`, and on the same
// address naming its caller, `mesh.seat.node-engine.call.<verb>.<node>.<caller>`, whose last token the bus lets each user
// publish with its own name alone (novox/hq issue 365): the caller is that token, and a call on the subject naming none is
// from an unproven caller, recorded so. The body is the call's arguments; the answer is {result, error, node}, as every
// tool's (mesh-tools node-tools). One service, `node-engine`, answers discovery with each verb's seat, tool and node.

// EngineSeat is the seat.
const EngineSeat = catalogue.EngineSeat

// EngineToolSubject is where a verb of this node's seat is asked naming no caller.
func EngineToolSubject(verb, node string) string {
	return "mesh.seat." + EngineSeat + ".tool." + verb + "." + node
}

// EngineCallPattern is where a verb of this node's seat is asked naming its caller, any caller.
func EngineCallPattern(verb, node string) string {
	return "mesh.seat." + EngineSeat + ".call." + verb + "." + node + ".*"
}

// EngineEventSubject is where the seat says one of its events about this node.
func EngineEventSubject(event, node string) string {
	return "mesh.seat." + EngineSeat + ".event." + event + "." + node
}

// EngineCall is one call of the seat: its verb, its caller as the bus proved it, and its arguments.
type EngineCall struct {
	Verb   string
	Caller string
	Args   map[string]any
}

// EngineFunc answers one call: its result, or why it is refused.
type EngineFunc func(EngineCall) (any, error)

// CallerOfSubject is the caller a call's subject names for this node's verb: the last token of a caller-named subject,
// each `~` a dot, or stdio.Unproven on the subject that names none or on any other.
func CallerOfSubject(subject, verb, node string) string {
	token, ok := strings.CutPrefix(subject, "mesh.seat."+EngineSeat+".call."+verb+"."+node+".")
	if !ok || token == "" || strings.ContainsAny(token, ".*> ") {
		return stdio.Unproven
	}
	return strings.ReplaceAll(token, "~", ".")
}

// MaxEngineCall bounds a call's arguments: a path, an access and a few words.
const MaxEngineCall = 16 << 10

// AnswerEngineCall is the envelope one call is answered with: the verb's result, or its refusal in words.
func AnswerEngineCall(handle EngineFunc, verb, node, subject string, body []byte) []byte {
	envelope := map[string]any{"node": node}
	say := func() []byte {
		raw, err := json.Marshal(envelope)
		if err != nil {
			raw, _ = json.Marshal(map[string]any{"node": node, "error": "the answer could not be written: " + err.Error()})
		}
		return raw
	}
	args := map[string]any{}
	switch {
	case handle == nil:
		envelope["error"] = "this engine serves no verb of " + EngineSeat
		return say()
	case len(body) > MaxEngineCall:
		envelope["error"] = fmt.Sprintf("the call is %d bytes, more than a call of %s takes", len(body), EngineSeat)
		return say()
	case len(strings.TrimSpace(string(body))) > 0:
		if err := json.Unmarshal(body, &args); err != nil {
			envelope["error"] = "the call's arguments are not an object: " + err.Error()
			return say()
		}
	}
	result, err := handle(EngineCall{Verb: verb, Caller: CallerOfSubject(subject, verb, node), Args: args})
	if err != nil {
		envelope["error"] = err.Error()
	} else {
		envelope["result"] = result
	}
	return say()
}

// EngineSeats is a link that serves this node's seat. Optional, as Asked is.
type EngineSeats interface {
	ServeEngineSeat(EngineFunc)
}

// ServeEngineSeat answers every verb of the seat on this node's subjects, as the service node-engine, which says each
// verb's seat, tool and node to discovery (novox/hq ADR 0197). A bus whose user list is older than the grant refuses the
// subscriptions, said by the client; a caller then hears nobody answer.
func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
	if handle == nil || l.conn == nil {
		return
	}
	svc, err := micro.AddService(l.conn, micro.Config{Name: EngineSeat, Version: "1.0.0",
		Description: "the node seat node-engine of " + l.node + " (novox/hq ADR 0563)", QueueGroupDisabled: true})
	if err != nil {
		return
	}
	for _, v := range catalogue.EngineVerbs() {
		verb := v.Name
		answer := micro.HandlerFunc(func(r micro.Request) {
			_ = r.Respond(AnswerEngineCall(handle, verb, l.node, r.Subject(), r.Data()))
		})
		meta := map[string]string{"seat": EngineSeat, "tool": verb, "node": l.node}
		_ = svc.AddEndpoint(verb, answer, micro.WithEndpointSubject(EngineToolSubject(verb, l.node)),
			micro.WithEndpointMetadata(meta))
		_ = svc.AddEndpoint(verb+"-called", answer, micro.WithEndpointSubject(EngineCallPattern(verb, l.node)),
			micro.WithEndpointMetadata(meta))
	}
	l.engine = svc
}

// SayEngine says one of the seat's events about this node, on the events stream.
func (l *natsLink) SayEngine(event string, body any) error {
	raw, err := json.Marshal(body)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	_, err = l.js.Publish(EngineEventSubject(event, l.node), raw, nats.Context(ctx))
	return err
}

// The engine asks the operator under its own asker name (asks.EngineAsker) on the router's seat (ADR 0259 §3, as ADR
// 0563 §3 changes it): its ask and cancel on the seat's work stream, its warrants heard on its own decided subject, and
// its asks' state read from the router's records under its own name.

// RouterRecords is the router's bucket of asks, which each asker reads under its own name: the state `asks` of the
// operator-channel's holder, the messenger (mesh-catalog modules/messenger).
const RouterRecords = "messenger_asks"

// OperatorAsks is a link that asks the operator as this node's engine. Optional.
type OperatorAsks interface {
	PublishAsk(subject string, body any, id string) error
	RouterRecord(key string) (*stdio.StateEntry, error)
	HearWarrants(asker string, heard func(asks.Warrant)) error
}

// PublishAsk publishes an ask or a cancel on the router's seat, once per id within the stream's duplicate window.
func (l *natsLink) PublishAsk(subject string, body any, id string) error {
	raw, err := json.Marshal(body)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	_, err = l.js.Publish(subject, raw, nats.MsgId(id), nats.Context(ctx))
	return err
}

// RouterRecord reads one key of the router's records, `<asker>.<id>`; nil when it holds none.
func (l *natsLink) RouterRecord(key string) (*stdio.StateEntry, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	msg, err := l.conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+RouterRecords+".$KV."+RouterRecords+"."+key, nil)
	if err != nil {
		return nil, err
	}
	if status := msg.Header.Get("Status"); status == "404" {
		return nil, nil
	} else if status != "" {
		return nil, fmt.Errorf("the router's record %s could not be read: %s %s", key, status, msg.Header.Get("Description"))
	}
	return &stdio.StateEntry{Key: key, Value: msg.Data}, nil
}

// HearWarrants hears the warrants for asker on its decided subject, each handed to heard as it arrives.
func (l *natsLink) HearWarrants(asker string, heard func(asks.Warrant)) error {
	sub, err := l.conn.Subscribe(asks.DecidedSubject(asker), func(msg *nats.Msg) {
		var w asks.Warrant
		if json.Unmarshal(msg.Data, &w) == nil {
			heard(w)
		}
	})
	if err != nil {
		return err
	}
	l.warrants = sub
	return nil
}

// errNoAsking is a link that cannot ask the operator.
var errNoAsking = errors.New("the link open now cannot ask the operator")

// discoveryQuiet is how long the engine waits after the last answer to the bus's discovery, and discoveryMost the
// longest it waits in all.
const (
	discoveryQuiet = 300 * time.Millisecond
	discoveryMost  = 2 * time.Second
)

// HeardServed asks the bus's discovery (novox/hq ADR 0197) whether a seat's verb is answered on this machine: an
// endpoint naming the seat, the verb as its tool and this node.
func (l *natsLink) HeardServed(seat, verb string) (bool, error) {
	inbox := l.conn.NewRespInbox()
	sub, err := l.conn.SubscribeSync(inbox)
	if err != nil {
		return false, err
	}
	defer func() { _ = sub.Unsubscribe() }()
	if err := l.conn.PublishRequest("$SRV.INFO", inbox, nil); err != nil {
		return false, fmt.Errorf("asking the bus who serves what: %w", err)
	}
	heard := false
	deadline := time.Now().Add(discoveryMost)
	for time.Now().Before(deadline) {
		msg, err := sub.NextMsg(discoveryQuiet)
		if err != nil {
			break
		}
		var info micro.Info
		if json.Unmarshal(msg.Data, &info) != nil {
			continue
		}
		for _, e := range info.Endpoints {
			node := e.Metadata["node"]
			if node == "" {
				node = info.ID
			}
			if e.Metadata["seat"] == seat && e.Metadata["tool"] == verb && node == l.node {
				heard = true
			}
		}
	}
	return heard, nil
}

// HeardServed is the link open now asking the bus's discovery whether a seat's verb is answered on this machine.
func (q *Session) HeardServed(seat, verb string) (bool, error) {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	asker, ok := bus.(interface {
		HeardServed(string, string) (bool, error)
	})
	if bus == nil || !ok {
		return false, fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked)
	}
	return asker.HeardServed(seat, verb)
}

// ---- the engine's acts in the bus's own storage (ADR 0563 §2, §3) -------------------------------------------------

// actsStore is this node's keys of ActsBucket over the link open now: bound, never created (the grants part creates
// it, as the node-engine's declared state).
func (l *natsLink) actsStore() (nats.KeyValue, error) {
	kv, err := l.js.KeyValue(ActsBucket)
	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; asked again once it is): %w", ActsBucket, err)
	}
	return kv, nil
}

// BusState is the engine's ActStore: its own node's keys of ActsBucket, `<node>.<key>`, over whichever link the session
// has open; with none open every read and write is refused, and nothing is kept anywhere else.
type BusState struct {
	Session *Session
	Node    string
}

func (b BusState) kv() (nats.KeyValue, error) {
	b.Session.init()
	b.Session.mu.Lock()
	bus := b.Session.bus
	b.Session.mu.Unlock()
	l, ok := bus.(interface{ actsStore() (nats.KeyValue, error) })
	if bus == nil || !ok {
		return nil, fmt.Errorf("%w: this machine has no link to the bus open now, where its acts are kept", ErrNotAsked)
	}
	return l.actsStore()
}

func (b BusState) key(k string) string { return b.Node + "." + k }

// changed is a write that lost its compare-and-set, as stdio.ErrChanged.
func changed(err error) error {
	var api *nats.APIError
	if errors.Is(err, nats.ErrKeyExists) || (errors.As(err, &api) && api.ErrorCode == nats.JSErrCodeStreamWrongLastSequence) {
		return stdio.ErrChanged
	}
	return err
}

func (b BusState) Get(key string) (*stdio.StateEntry, error) {
	kv, err := b.kv()
	if err != nil {
		return nil, err
	}
	e, err := kv.Get(b.key(key))
	if errors.Is(err, nats.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &stdio.StateEntry{Key: key, Value: e.Value(), Revision: e.Revision()}, nil
}

func (b BusState) Create(key string, value any) (uint64, error) {
	kv, err := b.kv()
	if err != nil {
		return 0, err
	}
	raw, err := json.Marshal(value)
	if err != nil {
		return 0, err
	}
	rev, err := kv.Create(b.key(key), raw)
	return rev, changed(err)
}

func (b BusState) Update(key string, value any, revision uint64) (uint64, error) {
	kv, err := b.kv()
	if err != nil {
		return 0, err
	}
	raw, err := json.Marshal(value)
	if err != nil {
		return 0, err
	}
	rev, err := kv.Update(b.key(key), raw, revision)
	return rev, changed(err)
}

func (b BusState) Delete(key string) error {
	kv, err := b.kv()
	if err != nil {
		return err
	}
	return kv.Delete(b.key(key))
}

// Keys are this node's keys under prefix, without the node: a watch of `<node>.>` alone, the one filter the node's
// grant names, its values left out.
func (b BusState) Keys(prefix string) ([]string, error) {
	kv, err := b.kv()
	if err != nil {
		return nil, err
	}
	w, err := kv.Watch(b.Node+".>", nats.MetaOnly(), nats.IgnoreDeletes())
	if err != nil {
		return nil, err
	}
	defer func() { _ = w.Stop() }()
	var out []string
	for {
		select {
		case e := <-w.Updates():
			if e == nil {
				sort.Strings(out)
				return out, nil
			}
			if k := strings.TrimPrefix(e.Key(), b.Node+"."); strings.HasPrefix(k, prefix) {
				out = append(out, k)
			}
		case <-time.After(10 * time.Second):
			return nil, errors.New("the node-engine's acts could not be listed within 10s")
		}
	}
}