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

package bus

import (
	"context"
	"encoding/json"
	"fmt"
	"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 and when it arrived.
type Heard struct {
	Subject string
	Body    json.RawMessage
	At      time.Time
}

// 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 said as an error naming the subject, at once;
// 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, err error) {
	if len(subjects) == 0 {
		return Heard{}, false, fmt.Errorf("a wait names at least one subject")
	}
	got := make(chan Heard, 1)
	refused := make(chan string, len(subjects))
	var subs []*nats.Subscription
	defer func() {
		for _, s := range subs {
			_ = s.Unsubscribe()
		}
		c.mu.Lock()
		for _, subject := range subjects {
			delete(c.waits[subject], refused)
			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 subjects {
		if c.waits[subject] == nil {
			c.waits[subject] = map[chan string]bool{}
		}
		c.waits[subject][refused] = true
	}
	c.mu.Unlock()
	for _, subject := range subjects {
		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()}:
			default:
			}
		})
		if err != nil {
			return Heard{}, false, 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, fmt.Errorf("the bus did not take the wait: %w", err)
	}
	timer := time.NewTimer(within)
	defer timer.Stop()
	select {
	case h := <-got:
		return h, true, nil
	case subject := <-refused:
		return Heard{}, false, fmt.Errorf("the bus refused this account a subscription on %s: it is not in the "+
			"account's wait grant (novox/hq ADR 0367 §2)", subject)
	case <-timer.C:
		return Heard{}, false, nil
	case <-ctx.Done():
		return Heard{}, false, 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], `"`)
}