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 }