mesh-tools/node-tools/internal/bus/wait.go

package bus

import (
	"context"
	"encoding/json"
	"fmt"
	"slices"
	"strings"
	"time"

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

// Waiting on the bus for one event (novox/hq ADR 0367): what the mesh MCP server's `mesh_wait` does.
//
// **A subscription for the call, and nothing kept** (§2): no durable consumer, no queue, no stream. It
// subscribes the subjects when called and unsubscribes them before it answers, so an event published
// between two calls is not heard — a session reads the state from its owner after every wake (§4).

// WaitIssued is one event family the mesh issued this account to wait on (ADR 0367 §2): the name a session
// asks for, and the subject it lands on. Only the tool runner's membership carries any.
type WaitIssued struct {
	Event   string `json:"event"`
	Subject string `json:"subject"`
}

// Heard is one event a wait heard: the subject it arrived on, its body, when it arrived, and when it was published, the
// event's own header `x-time` as its emitter wrote it, empty for an event that carries none (design 34 every-event §12).
type Heard struct {
	Subject   string
	Body      json.RawMessage
	At        time.Time
	Published string
}

// HearFirst subscribes the subjects for this call alone and answers the first message whose body match
// accepts (every one, when match is nil), or ok false once `within` has passed. Every subscription is
// closed before it returns. A subscription the bus refuses is noted and the others go on waiting; refused lists
// those noted by the time it answers (design 34 every-event §16). When the bus refused every subscription of the
// call, it still waits out `within` and then answers the refusal, so a caller looping on it asks once a timeout and
// never at once again. A caller gone — ctx done — ends the wait at once with ctx's error, so nothing waits on for
// nobody.
func (c *Conn) HearFirst(ctx context.Context, subjects []string, within time.Duration, match func(json.RawMessage) bool) (heard Heard, ok bool, refused []string, err error) {
	if len(subjects) == 0 {
		return Heard{}, false, nil, fmt.Errorf("a wait names at least one subject")
	}
	var unique []string
	for _, subject := range subjects {
		if !slices.Contains(unique, subject) {
			unique = append(unique, subject)
		}
	}
	got := make(chan Heard, 1)
	refusals := make(chan string, len(unique))
	var subs []*nats.Subscription
	defer func() {
		for _, s := range subs {
			_ = s.Unsubscribe()
		}
		c.mu.Lock()
		for _, subject := range unique {
			delete(c.waits[subject], refusals)
			if len(c.waits[subject]) == 0 {
				delete(c.waits, subject)
			}
		}
		c.mu.Unlock()
	}()
	c.mu.Lock()
	if c.waits == nil {
		c.waits = map[string]map[chan string]bool{}
	}
	for _, subject := range unique {
		if c.waits[subject] == nil {
			c.waits[subject] = map[chan string]bool{}
		}
		c.waits[subject][refusals] = true
	}
	c.mu.Unlock()
	for _, subject := range unique {
		sub, err := c.nc.Subscribe(subject, func(msg *nats.Msg) {
			body := json.RawMessage(msg.Data)
			if !json.Valid(body) {
				quoted, _ := json.Marshal(string(msg.Data))
				body = quoted
			}
			if match != nil && !match(body) {
				return
			}
			select {
			case got <- Heard{Subject: msg.Subject, Body: body, At: time.Now().UTC(), Published: msg.Header.Get("x-time")}:
			default:
			}
		})
		if err != nil {
			return Heard{}, false, nil, fmt.Errorf("could not wait on %s: %w", subject, err)
		}
		subs = append(subs, sub)
	}
	// The bus has every subscription — or has said which it refused — once this returns.
	if err := c.nc.FlushTimeout(5 * time.Second); err != nil {
		return Heard{}, false, nil, fmt.Errorf("the bus did not take the wait: %w", err)
	}
	timer := time.NewTimer(within)
	defer timer.Stop()
	for {
		select {
		case h := <-got:
			return h, true, refused, nil
		case subject := <-refusals:
			if !slices.Contains(refused, subject) {
				refused = append(refused, subject)
			}
		case <-timer.C:
			if len(refused) == len(unique) {
				return Heard{}, false, refused, fmt.Errorf("the bus refused this account a subscription on %s: the "+
					"grants part grants this machine's tool runner every event subject (novox/hq ADR 0670 §1), so its "+
					"grant is not composed yet; read the state from its owner meanwhile", strings.Join(refused, ", "))
			}
			return Heard{}, false, refused, nil
		case <-ctx.Done():
			return Heard{}, false, refused, ctx.Err()
		}
	}
}

// waitRefused tells every wait on a subject the bus refused that it was refused; true when one was waiting.
func (c *Conn) waitRefused(subject string) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	told := false
	for ch := range c.waits[subject] {
		select {
		case ch <- subject:
		default:
		}
		told = true
	}
	return told
}

// Subscriptions is how many subscriptions this connection holds now: what a wait leaves behind is none.
func (c *Conn) Subscriptions() int { return c.nc.NumSubscriptions() }

// subjectRefused is the subject a subscription refusal names, or "".
func subjectRefused(err error) string {
	m := refusedSubject.FindStringSubmatch(err.Error())
	if len(m) < 2 {
		return ""
	}
	return strings.Trim(m[1], `"`)
}