mesh-delivery/modules/mesh-delivery-view/internal/view/backlog.go

package view

import (
	"context"
	"errors"
	"fmt"
	"time"

	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
	"github.com/nats-io/nats.go/jetstream"
)

// The backlog of the build seats (novox/hq ADR 0525 §4, issue 1887 A8a): for each language and kind of ask, how many asks wait
// on the seat's stream for a node to pull them, how many a node holds without having finished them, and how old the oldest
// waiting ask is. It is read from the stream's shared pull worker (buildstep.AskWorker) when the view is asked: NumPending
// waits, NumAckPending is held by a node, and the oldest waiting ask is the stream's message at the worker's next sequence.

// BacklogRow is one language's kind of ask: Waiting asks not yet pulled, Held asks pulled by a node and not yet ended, and the
// oldest waiting ask's time and age in seconds. Error says why the row could not be read, never dropping it silently.
type BacklogRow struct {
	Language   string    `json:"language"`
	Kind       string    `json:"kind"`
	Waiting    uint64    `json:"waiting"`
	Held       uint64    `json:"held"`
	Oldest     time.Time `json:"oldest,omitzero"`
	OldestSecs int64     `json:"oldest-secs,omitempty"`
	Error      string    `json:"error,omitempty"`
}

// Backlog reads the seats' workers. Info is the worker's counts and the stream sequence it delivers next; Message is the body
// of the first message of the stream at or after seq on the subject.
type Backlog struct {
	Info    func(stream, worker string) (waiting, held, next uint64, err error)
	Message func(stream, subject string, seq uint64) ([]byte, error)
	Now     func() time.Time
}

// Rows are the rows of every language and kind that has an ask waiting or held, or that could not be read.
func (b *Backlog) Rows() []BacklogRow {
	now := time.Now
	if b.Now != nil {
		now = b.Now
	}
	out := []BacklogRow{}
	for _, lang := range buildstep.AskLanguages {
		stream, _ := buildstep.AskStream(lang)
		for _, kind := range buildstep.AskKinds {
			worker, _ := buildstep.AskWorker(lang, kind)
			subject, _ := buildstep.AskSubject(lang, kind)
			row := BacklogRow{Language: lang, Kind: kind}
			waiting, held, next, err := b.Info(stream, worker)
			if err != nil {
				row.Error = err.Error()
				out = append(out, row)
				continue
			}
			row.Waiting, row.Held = waiting, held
			if waiting > 0 {
				// Both forms of the accept subject: main's and the asker's (design 47 G §600, §484); the oldest of the two.
				var oldest *buildstep.StepAsk
				var err error
				for _, form := range []string{subject, subject + ".*"} {
					body, gerr := b.Message(stream, form, next)
					if errors.Is(gerr, jetstream.ErrMsgNotFound) {
						continue
					}
					if gerr != nil {
						err = gerr
						break
					}
					a, derr := buildstep.DecodeStepAsk(body)
					if derr != nil {
						err = derr
						break
					}
					if oldest == nil || a.At.Before(oldest.At) {
						oldest = &a
					}
				}
				if err == nil && oldest == nil {
					err = jetstream.ErrMsgNotFound
				}
				if err == nil {
					row.Oldest, row.OldestSecs = oldest.At, int64(now().Sub(oldest.At).Seconds())
				}
				if err != nil {
					row.Error = fmt.Sprintf("the oldest waiting ask cannot be read: %v", err)
				}
			}
			if row.Waiting > 0 || row.Held > 0 || row.Error != "" {
				out = append(out, row)
			}
		}
	}
	return out
}

// BacklogOver reads the backlog from the bus: the worker's consumer info, and the stream's message by sequence.
func BacklogOver(js jetstream.JetStream, timeout time.Duration) *Backlog {
	ctx := func() (context.Context, context.CancelFunc) {
		return context.WithTimeout(context.Background(), timeout)
	}
	return &Backlog{
		Info: func(stream, worker string) (uint64, uint64, uint64, error) {
			c, cancel := ctx()
			defer cancel()
			s, err := js.Stream(c, stream)
			if err != nil {
				return 0, 0, 0, fmt.Errorf("the stream %s: %w", stream, err)
			}
			cons, err := s.Consumer(c, worker)
			if err != nil {
				return 0, 0, 0, fmt.Errorf("the worker %s: %w", worker, err)
			}
			info := cons.CachedInfo()
			if i, err := cons.Info(c); err == nil {
				info = i
			}
			return uint64(info.NumPending), uint64(info.NumAckPending), info.Delivered.Stream + 1, nil
		},
		Message: func(stream, subject string, seq uint64) ([]byte, error) {
			c, cancel := ctx()
			defer cancel()
			s, err := js.Stream(c, stream)
			if err != nil {
				return nil, err
			}
			m, err := s.GetMsg(c, seq, jetstream.WithGetMsgSubject(subject))
			if err != nil {
				return nil, err
			}
			return m.Data, nil
		},
	}
}