mesh-host/internal/link/running.go

package link

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

	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
	"github.com/nats-io/nats.go"
)

// The node seat's event running, said by every node-engine for each module it saw running healthy, is kept on the stream
// SEAT_node-engine_running, the last message of each subject with no age limit (design 47 G §199, §260). The node reads it
// at each start through an ordered consumer it makes for that one read (§556), then follows a durable consumer of its own
// from its acknowledged position: what the other nodes run is known from the last message of each subject however long ago it
// was said, never from a have key or a call.

// RunningBus is a link that hears the seat's event running of every node.
type RunningBus interface {
	// FirstRunning calls first with the last message of every subject, through an ordered consumer made for this read, and
	// answers when it has read them all.
	FirstRunning(ctx context.Context, first func([]byte)) error
	// HearRunning calls heard with each message after the durable consumer's acknowledged position (a new consumer begins at
	// the last message of each subject), until stop is called or the link ends.
	HearRunning(consumer string, heard func([]byte)) (stop func(), err error)
}

// RunningConsumer is the name of this node's durable consumer of the stream.
func RunningConsumer(node string) string { return "node-engine-" + node + "-running" }

func (l *natsLink) FirstRunning(ctx context.Context, first func([]byte)) error {
	sub, err := l.js.SubscribeSync(nodeengine.SubjectRunningAll, nats.OrderedConsumer(), nats.DeliverLastPerSubject(),
		nats.BindStream(nodeengine.StreamRunning))
	if err != nil {
		return fmt.Errorf("the stream %s cannot be read: %w", nodeengine.StreamRunning, err)
	}
	defer sub.Unsubscribe()
	info, err := sub.ConsumerInfo()
	if err != nil {
		return fmt.Errorf("the stream %s cannot be read: %w", nodeengine.StreamRunning, err)
	}
	for total, got := info.Delivered.Consumer+info.NumPending, uint64(0); got < total; got++ {
		msg, err := sub.NextMsgWithContext(ctx)
		if err != nil {
			return fmt.Errorf("reading the last running of each subject (%d of %d): %w", got, total, err)
		}
		first(msg.Data)
	}
	return nil
}

func (l *natsLink) HearRunning(consumer string, heard func([]byte)) (func(), error) {
	sub, err := l.js.Subscribe(nodeengine.SubjectRunningAll, func(m *nats.Msg) {
		heard(m.Data)
		_ = m.Ack()
	}, nats.Durable(consumer), nats.BindStream(nodeengine.StreamRunning), nats.DeliverLastPerSubject(), nats.AckExplicit(),
		nats.ManualAck())
	if err != nil {
		return nil, fmt.Errorf("the stream %s cannot be followed: %w", nodeengine.StreamRunning, err)
	}
	return func() { _ = sub.Unsubscribe() }, nil
}

var _ RunningBus = (*natsLink)(nil)

// FollowRunning hands every node's running event to heard: the last of each subject first, with readEnded called after them
// (until then the caller acts on none, §556), then each after them, until ctx ends. An error is the stream or the link not
// being there, which the caller tries again.
func (q *Session) FollowRunning(ctx context.Context, heard func(nodeengine.Running), readEnded func()) error {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	h, ok := bus.(RunningBus)
	if bus == nil || !ok {
		return ErrNoLink
	}
	decode := func(raw []byte) {
		var ev nodeengine.Running
		if json.Unmarshal(raw, &ev) == nil && ev.Node != "" && ev.Module != "" {
			heard(ev)
		}
	}
	if err := h.FirstRunning(ctx, decode); err != nil {
		return err
	}
	stop, err := h.HearRunning(RunningConsumer(q.Membership.Node), decode)
	if err != nil {
		return err
	}
	defer stop()
	readEnded()
	<-ctx.Done()
	return nil
}

// OverNATS reads the stream as the link's session does, over its connection.
func (b OverNATS) FirstRunning(ctx context.Context, first func([]byte)) error {
	return (&natsLink{js: b.JS}).FirstRunning(ctx, first)
}

// HearRunning follows the stream over the connection (see natsLink.HearRunning).
func (b OverNATS) HearRunning(consumer string, heard func([]byte)) (func(), error) {
	return (&natsLink{js: b.JS}).HearRunning(consumer, heard)
}

var _ RunningBus = OverNATS{}