mesh-delivery/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go

package main

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

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

	"git.novox.be/novox/mesh-sdk/go/delivery"

	"mesh-delivery-builder/internal/build"
)

// heldOnTheBus says whether an ask a build key names is on its build seat's stream, or a build node wrote a step for it
// (issue 1910), read once at the take-over of a key at start; and, at the builder's check bound, whether an ask can still
// be answered (issue 1945). Never polled.
type heldOnTheBus struct{ js jetstream.JetStream }

// Held is build.Builder's Held: a step key `<node>.<ask id>` in the build module's steps, or a message of the stream under
// the ask's subject whose Nats-Msg-Id is the ask's id.
func (h heldOnTheBus) Held(ctx context.Context, a delivery.StepAsk) (bool, error) {
	seen, err := h.seen(ctx, a, false)
	if err != nil {
		return false, err
	}
	return seen.Step || seen.OnStream, nil
}

// Gone is build.Builder's Gone: why the ask can no longer be answered (build.AskSeen.Why), or "".
func (h heldOnTheBus) Gone(ctx context.Context, a delivery.StepAsk) (string, error) {
	seen, err := h.seen(ctx, a, true)
	if err != nil {
		return "", err
	}
	return seen.Why(), nil
}

// maxOnSubject bounds the messages of one subject a read walks (each a read by sequence, next by subject), and maxSteps
// the steps it reads whole to find the last ask taken: past them the read says it cannot tell, and the ask is waited for.
const (
	maxOnSubject = 5000
	maxSteps     = 200
)

// seen reads what the bus holds of the ask; with later, the ask asked last that a build node took too (AskSeen.Last), when
// the ask is on its stream.
func (h heldOnTheBus) seen(ctx context.Context, a delivery.StepAsk, later bool) (build.AskSeen, error) {
	seen := build.AskSeen{Kind: a.Kind, At: build.AskedAt(a.ID)}
	module, err := delivery.BuildModuleOf(a.Language)
	if err != nil {
		return seen, err
	}
	steps, err := h.js.KeyValue(ctx, delivery.StepsBucket(module))
	if err != nil {
		return seen, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
	}
	pattern, err := delivery.AskStepWatch(a.ID)
	if err != nil {
		return seen, err
	}
	keys, err := steps.ListKeysFiltered(ctx, pattern)
	if err != nil {
		return seen, err
	}
	for range keys.Keys() {
		keys.Stop()
		seen.Step = true // a node took it
		return seen, nil
	}
	stream, name, subject, err := h.stream(ctx, a.Language, a.Kind)
	seen.Stream = name
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		return seen, nil
	}
	if err != nil {
		return seen, err
	}
	var ahead []build.Queued
	err = onSubject(ctx, stream, subject, func(q build.Queued) bool {
		if q.Ask == a.ID {
			seen.OnStream, seen.PutAt = true, q.PutAt
			return false
		}
		ahead = append(ahead, q)
		return true
	})
	if err != nil || !seen.OnStream || !later {
		return seen, err
	}
	taken, err := stepKeys(ctx, steps)
	if err != nil {
		return seen, err
	}
	for i := range ahead {
		ahead[i].Step = stepState(ctx, steps, taken[ahead[i].Ask])
	}
	seen.Ahead = ahead
	since := seen.At
	if since.IsZero() {
		since = seen.PutAt
	}
	for _, q := range ahead {
		if q.At.Before(since) {
			since = q.At
		}
	}
	seen.Last = lastTaken(ctx, steps, taken, a.Kind, since)
	return seen, nil
}

// stream is the build seat's stream of language, its name and the subject of kind on it.
func (h heldOnTheBus) stream(ctx context.Context, language, kind string) (jetstream.Stream, string, string, error) {
	subject, err := delivery.AskSubject(language, kind)
	if err != nil {
		return nil, "", "", err
	}
	// The seat's stream by its name (delivery.AskStream), as the broker grants the builder its information and its
	// messages by sequence and nothing wider (issue 1910).
	name, err := delivery.AskStream(language)
	if err != nil {
		return nil, "", "", err
	}
	stream, err := h.js.Stream(ctx, name)
	return stream, name, subject, err
}

// onSubject hands each message of subject on stream to each, in order, until each answers false or the subject's
// messages end. A seat's stream is a work queue, which takes no second consumer over its subjects: its messages are read
// by sequence instead, each the next of the subject (one read per message, none for the sequences taken between), and
// none consumed.
func onSubject(ctx context.Context, stream jetstream.Stream, subject string, each func(build.Queued) bool) error {
	for seq, n := uint64(1), 0; ; n++ {
		if n >= maxOnSubject {
			return fmt.Errorf("%s holds more than %d messages before the one asked after: not read further", subject, maxOnSubject)
		}
		m, err := stream.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(subject))
		if errors.Is(err, jetstream.ErrMsgNotFound) {
			return nil
		}
		if err != nil {
			return err
		}
		q := build.Queued{Seq: m.Sequence, Ask: m.Header.Get(jetstream.MsgIDHeader), PutAt: m.Time}
		if _, kind, err := delivery.ParseAskSubject(m.Subject); err == nil {
			q.Kind = kind
		}
		if q.At = build.AskedAt(q.Ask); q.At.IsZero() {
			q.At = q.PutAt
		}
		if !each(q) {
			return nil
		}
		seq = m.Sequence + 1
	}
}

// stepKeys are the build module's step keys by ask id: `<node>.<ask id>` (a node's own key, such as its pause, has no
// ask id and is left out).
func stepKeys(ctx context.Context, steps jetstream.KeyValue) (map[string][]string, error) {
	all, err := steps.ListKeys(ctx)
	if err != nil {
		if errors.Is(err, jetstream.ErrNoKeysFound) {
			return map[string][]string{}, nil
		}
		return nil, err
	}
	defer all.Stop()
	out := map[string][]string{}
	for k := range all.Keys() {
		if _, id, ok := strings.Cut(k, "."); ok && !build.AskedAt(id).IsZero() {
			out[id] = append(out[id], k)
		}
	}
	return out, nil
}

// stepState is the state of the first of keys a node wrote, "" for none.
func stepState(ctx context.Context, steps jetstream.KeyValue, keys []string) string {
	for _, k := range keys {
		e, err := steps.Get(ctx, k)
		if err != nil {
			continue
		}
		if s, err := delivery.DecodeStep(e.Value()); err == nil {
			return s.State
		}
	}
	return ""
}

// lastTaken is the ask asked last after since that a build node took, of a kind whose taking proves kind passed
// (build.Proves), read newest first and at most maxSteps of them; nil for none. A step's tool is "<kind> ask"
// (mesh-sdk buildstep).
func lastTaken(ctx context.Context, steps jetstream.KeyValue, taken map[string][]string, kind string, since time.Time) *build.Taken {
	var ids []string
	for id := range taken {
		if build.AskedAt(id).After(since) {
			ids = append(ids, id)
		}
	}
	sort.Slice(ids, func(i, j int) bool { return build.AskedAt(ids[i]).After(build.AskedAt(ids[j])) })
	read := 0
	for _, id := range ids {
		for _, k := range taken[id] {
			if read >= maxSteps {
				return nil
			}
			read++
			e, err := steps.Get(ctx, k)
			if err != nil {
				continue
			}
			s, err := delivery.DecodeStep(e.Value())
			if err != nil {
				continue
			}
			stepKind, ok := strings.CutSuffix(s.Tool, " ask")
			if !ok || !build.Proves(stepKind, kind) {
				continue
			}
			return &build.Taken{Ask: id, Kind: stepKind, At: build.AskedAt(id), Node: s.Node, Started: s.Started}
		}
	}
	return nil
}

// Drop is build.Dropper's Drop: it removes the ask's message from its build seat's stream, found by its Nats-Msg-Id on its
// subject, when the stream still holds it. The builder's account must be granted $JS.API.STREAM.MSG.DELETE.<stream>.
func (h heldOnTheBus) Drop(ctx context.Context, a delivery.StepAsk) error {
	stream, name, subject, err := h.stream(ctx, a.Language, a.Kind)
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		return nil
	}
	if err != nil {
		return err
	}
	var seq uint64
	if err := onSubject(ctx, stream, subject, func(q build.Queued) bool {
		if q.Ask == a.ID {
			seq = q.Seq
			return false
		}
		return true
	}); err != nil || seq == 0 {
		return err
	}
	if err := stream.DeleteMsg(ctx, seq); err != nil && !errors.Is(err, jetstream.ErrMsgNotFound) {
		return fmt.Errorf("deleting message %d of %s (is $JS.API.STREAM.MSG.DELETE.%s granted to the builder?): %w", seq, name, name, err)
	}
	return nil
}

// Queue is what the tool builder_seat_queue answers (build.SeatQueue): the messages of language's build seat's stream, of
// every kind, oldest first and at most limit of each kind, each with its step's state and whether it still waits.
func (h heldOnTheBus) Queue(ctx context.Context, language string, limit int) (build.SeatQueue, error) {
	q := build.SeatQueue{Language: language, Note: build.QueueNote}
	module, err := delivery.BuildModuleOf(language)
	if err != nil {
		return q, err
	}
	stream, name, _, err := h.stream(ctx, language, delivery.AskDelivery)
	q.Stream = name
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		q.Note = name + " does not exist"
		return q, nil
	}
	if err != nil {
		return q, err
	}
	info, err := stream.Info(ctx)
	if err != nil {
		return q, err
	}
	q.Messages, q.FirstSeq, q.LastSeq = info.State.Msgs, info.State.FirstSeq, info.State.LastSeq
	steps, err := h.js.KeyValue(ctx, delivery.StepsBucket(module))
	if err != nil {
		return q, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
	}
	taken, err := stepKeys(ctx, steps)
	if err != nil {
		return q, err
	}
	for _, kind := range delivery.AskKinds {
		subject, _ := delivery.AskSubject(language, kind)
		var ahead []build.Queued
		if err := onSubject(ctx, stream, subject, func(m build.Queued) bool {
			ahead = append(ahead, m)
			return len(ahead) < limit
		}); err != nil {
			return q, err
		}
		if len(ahead) == 0 {
			continue
		}
		for i := range ahead {
			ahead[i].Step = stepState(ctx, steps, taken[ahead[i].Ask])
		}
		seen := build.AskSeen{Kind: kind, Ahead: ahead, Last: lastTaken(ctx, steps, taken, kind, ahead[0].At)}
		q.Kinds = append(q.Kinds, build.NewKindQueue(kind, seen, time.Now()))
	}
	return q, nil
}