mesh-controller/internal/link/seatask.go

package link

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

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

	"github.com/novox/mesh-controller/internal/broker"
)

// Asking a seat's holder one of its verbs, on the bus (novox/hq design 33 §4, ADR 0159, ADR 0239). It lived beside the
// build queue until the queue went (R4g, ADR 0525 clause 6).

// NodeSeatToolSubject is where a node-scoped 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 NodeSeatToolSubject(seat, verb, node string) string {
	return SeatToolSubject(seat, verb) + "." + node
}

// AskSeatTool asks one machine's holder of a node seat one of its verbs and reads its answer.
func AskSeatTool(ctx context.Context, conn *nats.Conn, seat, verb, node string, args any,
	timeout time.Duration) (Answer, error) {
	body, err := json.Marshal(args)
	if err != nil {
		return Answer{}, err
	}
	asking, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	subject := NodeSeatToolSubject(seat, verb, node)
	// **A refused question is known at once** (novox/hq to-be 45, D8 live on 2026-10-06): the bus
	// says it to the connection and the request would otherwise wait out its whole timeout, reading
	// as a machine that did not answer.
	refused, stop := refusalsOf(conn, subject)
	defer stop()
	type replied struct {
		msg *nats.Msg
		err error
	}
	done := make(chan replied, 1)
	go func() {
		msg, err := conn.RequestWithContext(asking, subject, body)
		done <- replied{msg, err}
	}()
	var reply *nats.Msg
	select {
	case r := <-done:
		reply, err = r.msg, r.err
	case why := <-refused:
		cancel()
		return Answer{}, fmt.Errorf("the bus refused the controller asking %s.%s of %s — its grants do not "+
			"name %s: %v", seat, verb, node, subject, why)
	}
	switch {
	case errors.Is(err, nats.ErrNoResponders):
		return Answer{}, fmt.Errorf("nothing on %s answers %s.%s: its holder is not running, or is "+
			"older than the verb", node, seat, verb)
	case errors.Is(err, context.DeadlineExceeded), errors.Is(err, nats.ErrTimeout):
		return Answer{}, fmt.Errorf("%s did not answer %s.%s within %s", node, seat, verb, timeout)
	case err != nil:
		return Answer{}, err
	}
	data, err := Whole(ctx, conn, subject, reply.Data)
	if err != nil {
		return Answer{}, fmt.Errorf("%s answered %s.%s too large for one message: %w", node, seat, verb, err)
	}
	var answer Answer
	if err := json.Unmarshal(data, &answer); err != nil {
		return Answer{}, fmt.Errorf("%s answered %s.%s with something unreadable: %w", node, seat, verb, err)
	}
	return answer, nil
}

// AskMeshSeatTool asks the holder of a mesh-scoped seat one of its verbs, on the seat's flat subject
// (design 33 §4: a mesh seat's verb carries no machine), and reads its answer (novox/hq ADR 0239: the
// delivery's owner's `stalled` and `close`).
func AskMeshSeatTool(ctx context.Context, conn *nats.Conn, seat, verb string, args any, timeout time.Duration) (Answer, error) {
	return askMeshSeatToolAt(ctx, conn, SeatToolSubject(seat, verb), seat, verb, args, timeout)
}

// AskMeshSeatToolAs is AskMeshSeatTool in the caller's own name: on the subject that names its caller (novox/hq
// issue 365), which the bus lets only that user publish, so the holder knows who asks from the subject alone —
// how the vault takes a value from the controller and nobody else (design 57's V2, issue 675).
func AskMeshSeatToolAs(ctx context.Context, conn *nats.Conn, caller, seat, verb string, args any,
	timeout time.Duration) (Answer, error) {
	subject := broker.CalledSubject(SeatToolSubject(seat, verb), caller)
	if subject == "" {
		return Answer{}, fmt.Errorf("%q cannot ask %s.%s in its own name: it is not a bus user's name", caller, seat, verb)
	}
	return askMeshSeatToolAt(ctx, conn, subject, seat, verb, args, timeout)
}

func askMeshSeatToolAt(ctx context.Context, conn *nats.Conn, subject, seat, verb string, args any,
	timeout time.Duration) (Answer, error) {
	body, err := json.Marshal(args)
	if err != nil {
		return Answer{}, err
	}
	asking, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	refused, stop := refusalsOf(conn, subject)
	defer stop()
	type replied struct {
		msg *nats.Msg
		err error
	}
	done := make(chan replied, 1)
	go func() {
		msg, err := conn.RequestWithContext(asking, subject, body)
		done <- replied{msg, err}
	}()
	var reply *nats.Msg
	select {
	case r := <-done:
		reply, err = r.msg, r.err
	case why := <-refused:
		cancel()
		return Answer{}, fmt.Errorf("%w: the bus refused the controller asking %s.%s — its grants do not name %s: %v",
			ErrGrantRefused, seat, verb, subject, why)
	}
	switch {
	case errors.Is(err, nats.ErrNoResponders):
		return Answer{}, fmt.Errorf("%w: nothing answers %s.%s — its holder is not running, or is older than the verb",
			ErrNothingServes, seat, verb)
	case errors.Is(err, context.DeadlineExceeded), errors.Is(err, nats.ErrTimeout):
		return Answer{}, fmt.Errorf("%s did not answer %s within %s", seat, verb, timeout)
	case err != nil:
		return Answer{}, err
	}
	data, err := Whole(ctx, conn, subject, reply.Data)
	if err != nil {
		return Answer{}, fmt.Errorf("%s answered %s too large for one message: %w", seat, verb, err)
	}
	var answer Answer
	if err := json.Unmarshal(data, &answer); err != nil {
		return Answer{}, fmt.Errorf("%s answered %s with something unreadable: %w", seat, verb, err)
	}
	return answer, nil
}