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

package main

import (
	"context"
	"errors"
	"fmt"

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

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

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

// bucketOpener is the part of the bus the watches use.
type bucketOpener interface {
	KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error)
}

// kvStore is the builds bucket as the builder's store: create and compare-and-set.
type kvStore struct{ kv jetstream.KeyValue }

func conflict(err error) error {
	if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
		return build.ErrConflict
	}
	return err
}

func (s kvStore) Get(ctx context.Context, key string) ([]byte, uint64, bool, error) {
	e, err := s.kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, 0, false, nil
	}
	if err != nil {
		return nil, 0, false, err
	}
	return e.Value(), e.Revision(), true, nil
}

func (s kvStore) Create(ctx context.Context, key string, value []byte) (uint64, error) {
	rev, err := s.kv.Create(ctx, key, value)
	return rev, conflict(err)
}

func (s kvStore) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) {
	rev, err := s.kv.Update(ctx, key, value, revision)
	return rev, conflict(err)
}

func (s kvStore) Keys(ctx context.Context) ([]string, error) {
	keys, err := s.kv.Keys(ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil
	}
	return keys, err
}

func opOf(e jetstream.KeyValueEntry) bool { return e.Operation() == jetstream.KeyValuePut }

// settle reads the entries a watcher holds now, up to its mark of the last values, and returns the watcher for the
// changes that follow.
func settle(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyValueEntry)) (jetstream.KeyWatcher, error) {
	w, err := kv.WatchAll(ctx)
	if err != nil {
		return nil, err
	}
	for {
		select {
		case <-ctx.Done():
			w.Stop()
			return nil, ctx.Err()
		case e, ok := <-w.Updates():
			if !ok {
				return nil, errors.New("the watch ended")
			}
			if e == nil {
				return w, nil
			}
			each(e)
		}
	}
}

// follow hands each change of a watcher to each until ctx ends; a watch that ends is an error, and the process manager
// starts the program again.
func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.KeyValueEntry)) error {
	defer w.Stop()
	for {
		select {
		case <-ctx.Done():
			return nil
		case e, ok := <-w.Updates():
			if !ok {
				return errors.New("a watch ended")
			}
			if e != nil {
				each(e)
			}
		}
	}
}

// stepsWaiter waits for a build module's step by watching its key in the module's steps bucket (ADR 0469 clause 2): the
// watch hands over the value the key holds now, so a step that ended before the watch began is read at once, and every
// change after. No timer: the caller bounds the wait with its context.
type stepsWaiter struct{ js bucketOpener }

func (w stepsWaiter) Wait(ctx context.Context, module, key string) (delivery.Step, error) {
	kv, err := w.js.KeyValue(ctx, delivery.StepsBucket(module))
	if err != nil {
		return delivery.Step{}, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
	}
	watch, err := kv.Watch(ctx, key)
	if err != nil {
		return delivery.Step{}, err
	}
	defer watch.Stop()
	for {
		select {
		case <-ctx.Done():
			return delivery.Step{}, ctx.Err()
		case e, ok := <-watch.Updates():
			if !ok {
				return delivery.Step{}, errors.New("the watch of the step ended")
			}
			if e == nil || !opOf(e) {
				continue
			}
			step, err := delivery.DecodeStep(e.Value())
			if err != nil {
				return delivery.Step{}, fmt.Errorf("the step %s is not a step: %w", key, err)
			}
			if step.State != delivery.StepRunning {
				return step, nil
			}
		}
	}
}

// manifestReader reads manifests.<digest> of the indexer and parses the manifest it holds.
type manifestReader struct{ kv jetstream.KeyValue }

func (m manifestReader) Manifest(ctx context.Context, digest string) (catalogue.Manifest, error) {
	key, err := delivery.ManifestsKey(digest)
	if err != nil {
		return catalogue.Manifest{}, err
	}
	e, err := m.kv.Get(ctx, key)
	if err != nil {
		return catalogue.Manifest{}, fmt.Errorf("manifests.%s: %w", key, err)
	}
	kept, err := delivery.DecodeManifest(e.Value())
	if err != nil {
		return catalogue.Manifest{}, err
	}
	return catalogue.ParseManifest([]byte(kept.JSON))
}