mesh-host/internal/link/running.go

package link

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

	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"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{}

// The builder's event build.ended (design 47 G A-293, as the assigner hears it, A-181, §556): the builder's stream of build
// ends keeps the last message of each module and fingerprint, and the node reads it whole at each start, then follows it, through
// one ordered consumer it makes for the run: what ended, and a failed build's words, are known from the stream however long ago
// they were said, never from the builder's builds. The engine's manifest consumes `mesh-delivery-builder.build.ended.**`.

// BuildEndsStream is the builder's stream of build ends as the bus names it, BuildEndsSubject every event it carries.
var (
	BuildEndsStream  = catalogue.StreamBusName(builds.BuildsStream)
	BuildEndsSubject = "mesh.stream.mesh-delivery-builder.event." + builds.BuildEndedEvent + ".>"
)

// BuildEndsBus is a link that hears the builder's build ends.
type BuildEndsBus interface {
	// HearBuildEnds calls heard with the last message of every subject, then readEnded once they are all read, then heard with
	// each after them, until ctx ends (nil) or the stream is lost (its error).
	HearBuildEnds(ctx context.Context, heard func([]byte), readEnded func()) error
}

func hearBuildEnds(ctx context.Context, js nats.JetStreamContext, heard func([]byte), readEnded func()) error {
	sub, err := js.SubscribeSync(BuildEndsSubject, nats.OrderedConsumer(), nats.DeliverLastPerSubject(), nats.BindStream(BuildEndsStream))
	if err != nil {
		return fmt.Errorf("the stream %s cannot be read: %w", BuildEndsStream, err)
	}
	defer sub.Unsubscribe()
	info, err := sub.ConsumerInfo()
	if err != nil {
		return fmt.Errorf("the stream %s cannot be read: %w", BuildEndsStream, err)
	}
	total := info.Delivered.Consumer + info.NumPending
	for got := uint64(0); ; got++ {
		if got == total {
			readEnded()
		}
		msg, err := sub.NextMsgWithContext(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return fmt.Errorf("the stream %s is not followed any more (%d read): %w", BuildEndsStream, got, err)
		}
		heard(msg.Data)
	}
}

func (l *natsLink) HearBuildEnds(ctx context.Context, heard func([]byte), readEnded func()) error {
	return hearBuildEnds(ctx, l.js, heard, readEnded)
}

// HearBuildEnds reads the stream as the link's session does, over its connection.
func (b OverNATS) HearBuildEnds(ctx context.Context, heard func([]byte), readEnded func()) error {
	return hearBuildEnds(ctx, b.JS, heard, readEnded)
}

var (
	_ BuildEndsBus = (*natsLink)(nil)
	_ BuildEndsBus = OverNATS{}
)

// FollowBuildEnds hands every build end the builder said to heard: the last of each module and fingerprint first, with
// readEnded called after them (until then the caller acts on none), then each after them, until ctx ends. A body that is no
// end of a module at a fingerprint is passed over. An error is the stream or the link not being there, which the caller tries
// again.
func (q *Session) FollowBuildEnds(ctx context.Context, heard func(builds.BuildEnded), readEnded func()) error {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	h, ok := bus.(BuildEndsBus)
	if bus == nil || !ok {
		return ErrNoLink
	}
	return h.HearBuildEnds(ctx, func(raw []byte) {
		var e builds.BuildEnded
		if json.Unmarshal(raw, &e) == nil && e.Module != "" && e.Fingerprint != "" {
			heard(e)
		}
	}, readEnded)
}