mesh-sdk/go/process/event_hub.go

package process

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"

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

// EventHub is the one pull of a module's durable consumer on the shared events stream (novox/hq ADR 0635 §5): a module
// has one durable, `<node>_<module>`, so two loops pulling it each take messages the other needs. Every part of the
// module that hears events registers a handler by subject pattern, and the hub pulls once and hands each message to the
// handler whose pattern it matches. A message no handler takes is acknowledged and said, never left to be redelivered.
// The hub never makes the stream or the consumer: they are the controller's, and a bus without them is an error of Start.
//
// Register every handler, then Start once: a message heard before its handler is registered would be taken by none.
type EventHub struct {
	logf func(string, ...any)

	mu     sync.RWMutex
	routes []hubRoute
	cons   jetstream.ConsumeContext
}

type hubRoute struct {
	pattern string
	handle  func(jetstream.Msg)
}

// NewEventHub is a hub saying through logf (nil says nothing).
func NewEventHub(logf func(string, ...any)) *EventHub {
	if logf == nil {
		logf = func(string, ...any) {}
	}
	return &EventHub{logf: logf}
}

// Handle registers handle for the subjects matching pattern (`*` one token, `>` the rest). The handler acknowledges
// the message itself. A pattern that overlaps one registered (one matches every subject of the other) is refused,
// so a message always has one taker.
func (h *EventHub) Handle(pattern string, handle func(jetstream.Msg)) error {
	if pattern == "" || handle == nil {
		return fmt.Errorf("an event handler needs a subject pattern and a handler")
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	for _, r := range h.routes {
		if subjectsOverlap(r.pattern, pattern) {
			return fmt.Errorf("the events %q overlap %q, already handled: a message has one taker", pattern, r.pattern)
		}
	}
	h.routes = append(h.routes, hubRoute{pattern, handle})
	return nil
}

// Patterns are the registered subject patterns, in order.
func (h *EventHub) Patterns() []string {
	h.mu.RLock()
	defer h.mu.RUnlock()
	out := make([]string, len(h.routes))
	for i, r := range h.routes {
		out[i] = r.pattern
	}
	return out
}

// Start binds the module's durable on the shared events stream and pulls it; a call after a start that took does
// nothing. It makes no stream and no consumer: a consumer the bus does not hold yet is the error of this call, and a
// later call (the module's next change of membership) binds it, so a part is not lost for the bus's order.
func (h *EventHub) Start(ctx context.Context, js jetstream.JetStream, cred Account) error {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.cons != nil {
		return nil
	}
	cons, err := cred.EventConsumer(ctx, js, EventsStream)
	if err == nil {
		h.cons, err = HearEvents(cons, h.hear, h.logf)
	}
	if err != nil {
		return fmt.Errorf("the consumer %s on %s is not heard: %w", cred.EventConsumerName(), EventsStream, err)
	}
	return nil
}

// StartUntilBound is Start that does not give up: when the module's consumer is not there yet it returns that error at
// once, and tries again on a goroutine of its own, after each wait, until it binds or ctx ends; then live is called (nil
// calls nothing), so a part that needs the events path live does what it does only then. wait returns false to stop
// trying (nil waits five seconds or for ctx). The bus makes the consumer when the module's account is composed, which
// says nothing a module can hear before the consumer exists, so the retry is a wait on that, bounded by ctx.
func (h *EventHub) StartUntilBound(ctx context.Context, js jetstream.JetStream, cred Account, wait func(context.Context) bool, live func()) error {
	err := h.Start(ctx, js, cred)
	if err == nil {
		if live != nil {
			live()
		}
		return nil
	}
	if wait == nil {
		wait = func(ctx context.Context) bool {
			t := time.NewTimer(5 * time.Second)
			defer t.Stop()
			select {
			case <-ctx.Done():
				return false
			case <-t.C:
				return true
			}
		}
	}
	go func() {
		for wait(ctx) && ctx.Err() == nil {
			if h.Start(ctx, js, cred) == nil {
				if live != nil {
					live()
				}
				return
			}
		}
	}()
	return err
}

// Stop ends the pull.
func (h *EventHub) Stop() {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.cons != nil {
		h.cons.Stop()
		h.cons = nil
	}
}

func (h *EventHub) hear(m jetstream.Msg) {
	h.mu.RLock()
	var take func(jetstream.Msg)
	for _, r := range h.routes {
		if subjectMatches(r.pattern, m.Subject()) {
			take = r.handle
			break
		}
	}
	h.mu.RUnlock()
	if take == nil {
		h.logf("events: %s is taken by no handler of this module, acknowledged", m.Subject())
		_ = m.Ack()
		return
	}
	take(m)
}

// subjectMatches says whether subject is one the pattern (`*` one token, `>` the rest) names.
func subjectMatches(pattern, subject string) bool {
	pt, st := strings.Split(pattern, "."), strings.Split(subject, ".")
	for i, t := range pt {
		if t == ">" {
			return len(st) > i
		}
		if i >= len(st) || (t != "*" && t != st[i]) {
			return false
		}
	}
	return len(pt) == len(st)
}

// subjectsOverlap says whether some subject matches both patterns.
func subjectsOverlap(a, b string) bool {
	at, bt := strings.Split(a, "."), strings.Split(b, ".")
	for i := 0; ; i++ {
		if i >= len(at) || i >= len(bt) {
			return len(at) == len(bt)
		}
		if at[i] == ">" || bt[i] == ">" {
			return true
		}
		if at[i] != "*" && bt[i] != "*" && at[i] != bt[i] {
			return false
		}
	}
}