mesh-sdk/go/process/serve.go

package process

import (
	"encoding/json"
	"fmt"
	"regexp"
	"sort"
	"strings"

	"github.com/nats-io/nats.go"

	"git.novox.be/novox/mesh-sdk/go/approval"
	"git.novox.be/novox/mesh-sdk/go/calladdress"
	"git.novox.be/novox/mesh-sdk/go/seatdef"
)

// SeatHandler answers one call of a seat's verb: caller is the bus user the subject the call arrived on names
// (the bus server enforces it, so it is a fact, never read from the body), or "" when the call came on the subject
// that names none (an unproven caller); body is the call's arguments. A returned error is the call's failure.
type SeatHandler func(caller string, body json.RawMessage) (any, error)

// callerUser is the bus user in a caller token: each `~` is a dot (mesh-tools node-tools/internal/bus/caller.go
// UserOfToken), "" for a token that is not a bus user's name.
var callerUser = regexp.MustCompile(`^[A-Za-z0-9_-]+(\.[A-Za-z0-9_-]+)*$`)

func userOfToken(token string) string {
	user := strings.ReplaceAll(token, "~", ".")
	if !callerUser.MatchString(user) {
		return ""
	}
	return user
}

// AnswerSeat answers the verbs of a mesh seat this process's module holds, over the bus, as the tool runner
// answers a bundle's (novox/hq design 47 G, lead answer A-107): on `mesh.seat.<seat>.call.<verb>.*`, handing the
// handler the caller its last token names, and on `mesh.seat.<seat>.tool.<verb>`, which names none. Every instance
// of the seat's holder subscribes in one queue group per seat, so a call reaches one of them. The reply is
// {result, error, node}, as Call reads it. The returned stop ends every subscription. Nothing is subscribed when
// any verb is malformed or a subscription is refused.
//
// Before a handler runs, the call's caller is checked against the verb's effective callers (design 47 G §610; lead
// answers A-165, A-179, A-186): definition answers the seat's definition from the seat list the process follows, false
// while it is not read. An unproven caller is refused, and so is every caller while the definition is unread, names no
// such verb or opens it to none; a proven caller is among them, since the grants part composes a verb's call subject
// only for its effective callers.
//
// A verb the definition marks `needs-approval` is held through the gate Gated gives (design 47 G §677; lead answers A-224,
// A-244): a call that carries the header approval.WaitHeader is kept and answered at once with the reply's field `held`,
// its end said on the seat's `gated-ended`; one that does not takes ADR 0286 §37's path. With no gate a marked verb is
// refused to every caller. With a gate the seat's verb `approvals` is served too, unless verbs answers it.
func (c *Conn) AnswerSeat(seat string, definition func() (seatdef.Definition, bool), verbs map[string]SeatHandler, opts ...SeatOption) (stop func(), err error) {
	var o seatOptions
	for _, f := range opts {
		f(&o)
	}
	if definition == nil {
		return nil, fmt.Errorf("AnswerSeat(%q) is given no definition of the seat to read its verbs' callers from", seat)
	}
	if len(verbs) == 0 {
		return nil, fmt.Errorf("AnswerSeat(%q) answers no verb", seat)
	}
	if o.gate != nil && verbs[seatdef.ApprovalsVerb] == nil {
		all := make(map[string]SeatHandler, len(verbs)+1)
		for v, h := range verbs {
			all[v] = h
		}
		gate := o.gate
		all[seatdef.ApprovalsVerb] = func(caller string, _ json.RawMessage) (any, error) { return gate.Approvals(caller) }
		verbs = all
	}
	names := make([]string, 0, len(verbs))
	for verb, h := range verbs {
		if h == nil {
			return nil, fmt.Errorf("AnswerSeat(%q): the verb %s has no handler", seat, verb)
		}
		if _, err := calladdress.Parse("seat:" + seat + "." + verb); err != nil {
			return nil, err
		}
		names = append(names, verb)
	}
	sort.Strings(names)
	queue := "serve.seat." + seat
	var subs []*nats.Subscription
	undo := func() {
		for _, s := range subs {
			_ = s.Unsubscribe()
		}
	}
	for _, verb := range names {
		if o.gate != nil {
			h := verbs[verb]
			o.gate.Serves(seat, verb, func(caller string, args json.RawMessage) (any, error) { return h(caller, args) })
		}
		h := c.admitted(seat, verb, definition, verbs[verb], o.gate)
		base := "mesh.seat." + seat
		for _, subject := range []string{base + ".call." + verb + ".*", base + ".tool." + verb} {
			subject := subject
			sub, err := c.nc.QueueSubscribe(subject, queue, func(msg *nats.Msg) { go c.answerCall(msg, subject, h) })
			if err != nil {
				undo()
				return nil, fmt.Errorf("answering %s: %w", subject, err)
			}
			subs = append(subs, sub)
		}
	}
	if err := c.nc.Flush(); err != nil {
		undo()
		return nil, err
	}
	return undo, nil
}

// SeatOption is an option of AnswerSeat.
type SeatOption func(*seatOptions)

type seatOptions struct{ gate *approval.Gate }

// Gated holds the seat's marked verbs through g, the holder's gate (design 47 G §677): Conn.Gate makes it.
func Gated(g *approval.Gate) SeatOption { return func(o *seatOptions) { o.gate = g } }

// heldHandler is a verb's handler behind its checks: its answer, the held answer of a kept call, or its error.
type heldHandler func(caller string, body json.RawMessage, wait bool) (any, *approval.Held, error)

// admitted is h behind the check of its caller against the verb's effective callers (A-179, A-186), and, for a verb the
// definition marks, behind the gate (§677).
func (c *Conn) admitted(seat, verb string, definition func() (seatdef.Definition, bool), h SeatHandler, gate *approval.Gate) heldHandler {
	return func(caller string, body json.RawMessage, wait bool) (any, *approval.Held, error) {
		if caller == "" {
			return nil, nil, fmt.Errorf("%s.%s takes no unproven caller: call it on its subject that names its caller, "+
				"mesh.seat.%s.call.%s.<caller>", seat, verb, seat, verb)
		}
		def, read := definition()
		if !read {
			return nil, nil, fmt.Errorf("the definition of the seat %s is not yet read by this process's seat list, so no caller of %s.%s "+
				"is admitted", seat, seat, verb)
		}
		for _, v := range def.Verbs {
			if v.Name != verb {
				continue
			}
			if len(v.EffectiveCallers()) == 0 {
				return nil, nil, fmt.Errorf("%s.%s is open to no caller", seat, verb)
			}
			if !v.NeedsApproval {
				answer, err := h(caller, body)
				return answer, nil, err
			}
			if gate == nil {
				return nil, nil, fmt.Errorf("Not run: %s.%s needs the operator's approval, and its holder serves it through no gate", seat, verb)
			}
			return gate.Hold(approval.Call{Address: seat + "." + verb, Module: c.account.Module, Seat: seat, Name: verb,
				Caller: caller, Args: body, Description: v.Description}, wait, func() (any, error) { return h(caller, body) })
		}
		return nil, nil, fmt.Errorf("the seat %s's definition names no verb %s, so no caller is admitted", seat, verb)
	}
}

// answerCall runs h for one call and replies {result, error, node}; a panic in h is the call's error, never silence.
func (c *Conn) answerCall(msg *nats.Msg, pattern string, h heldHandler) {
	caller := ""
	if strings.HasSuffix(pattern, ".*") {
		caller = userOfToken(msg.Subject[strings.LastIndex(msg.Subject, ".")+1:])
	}
	var reply struct {
		Result any    `json:"result,omitempty"`
		Error  string `json:"error,omitempty"`
		Node   string `json:"node,omitempty"`
		// Held is a kept call's answer (design 47 G §677; lead answer A-244).
		Held *approval.Held `json:"held,omitempty"`
	}
	wait := msg.Header != nil && msg.Header.Get(approval.WaitHeader) == approval.WaitHeld
	func() {
		defer func() {
			if r := recover(); r != nil {
				reply.Error = fmt.Sprintf("%s failed: %v", msg.Subject, r)
			}
		}()
		result, held, err := h(caller, json.RawMessage(msg.Data), wait)
		switch {
		case err != nil:
			reply.Error = err.Error()
		case held != nil:
			reply.Held = held
		case result == nil:
			reply.Result = json.RawMessage("null")
		default:
			reply.Result = result
		}
	}()
	reply.Node = c.account.Node
	body, err := json.Marshal(reply)
	if err != nil {
		body, _ = json.Marshal(map[string]string{"error": "the answer is not JSON: " + err.Error(), "node": c.account.Node})
	}
	_ = msg.Respond(body)
}