mesh-controller/internal/builderkeys/builderkeys.go

// Package builderkeys is how the controller reads what the builder and the indexer wrote (novox/hq ADR 0501, ADR 0498
// §8, ADR 0453 §5): a module's build at a fingerprint, in the builder's bucket; the indexer's trunk value of a module,
// which names that fingerprint; and a manifest by its digest. From R4e the controller asks no build but the builder's
// own module's: it takes every other module from the builder's done key, and it reads them here.
//
// It only reads. The builder writes its keys and the indexer its values; the controller's account reaches every
// bucket through the JetStream API it is granted (internal/broker: KindController).
package builderkeys

import (
	"context"
	"errors"
	"fmt"
	"strings"

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

// Key is a module's build at a fingerprint as the builder's bucket holds it. Found is false when there is no key (or
// no bucket yet): nothing built or queued that fingerprint. Revision is the key's revision, a new one for every
// rebuild of the same fingerprint (ADR 0429 §18).
type Key struct {
	Module      string
	Fingerprint string
	Found       bool
	Revision    uint64
	Build       delivery.BuilderBuild
}

// Done says the key is done.
func (k Key) Done() bool { return k.Found && k.Build.State == delivery.BuildDone }

// Failed says the key failed; its Error is the builder's own words (`digests differ:`, `not stored:`, …).
func (k Key) Failed() bool { return k.Found && k.Build.State == delivery.BuildFailed }

// Unreferenced are the artifacts of a done key that name no reference (ADR 0498 §8): a key done before R4e's
// delivery, or a candidate's image or package not yet built again for the trunk. Such a key is waited for, never
// delivered from.
func (k Key) Unreferenced() []string {
	var out []string
	for _, a := range k.Build.Artifacts {
		if strings.TrimSpace(a.Reference) == "" {
			out = append(out, a.Name)
		}
	}
	return out
}

// Says is the key's state in words: what a plan waiting on it says.
func (k Key) Says() string {
	if !k.Found {
		return fmt.Sprintf("no build of %s at %s yet", k.Module, short(k.Fingerprint))
	}
	switch k.Build.State {
	case delivery.BuildFailed:
		return fmt.Sprintf("the build of %s at %s failed: %s", k.Module, short(k.Fingerprint), firstLine(k.Build.Error))
	case delivery.BuildDone:
		if missing := k.Unreferenced(); len(missing) > 0 {
			return fmt.Sprintf("the build of %s at %s is done and waits for the reference of %s", k.Module,
				short(k.Fingerprint), strings.Join(missing, ", "))
		}
		return fmt.Sprintf("the build of %s at %s is done", k.Module, short(k.Fingerprint))
	}
	if k.Build.By != "" {
		return fmt.Sprintf("the build of %s at %s is %s on %s", k.Module, short(k.Fingerprint), k.Build.State, k.Build.By)
	}
	return fmt.Sprintf("the build of %s at %s is %s", k.Module, short(k.Fingerprint), k.Build.State)
}

// ErrNoValue is a module the indexer holds no value of at the commit asked: not written yet, or older than the
// bucket's history.
var ErrNoValue = errors.New("the indexer holds no value of the module at that commit")

// Reader reads the builder's and the indexer's buckets.
type Reader struct {
	JS jetstream.JetStream
}

// bucket is the named bucket, or nil when it is not there yet (the part that writes it not delivered).
func (r Reader) bucket(ctx context.Context, name string) (jetstream.KeyValue, error) {
	kv, err := r.JS.KeyValue(ctx, name)
	if errors.Is(err, jetstream.ErrBucketNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("the bucket %s: %w", name, err)
	}
	return kv, nil
}

// Build is the builder's key of module at fingerprint fp (`src2:` and 64 hex digits). A missing key, or a missing
// bucket, is a Key not Found and no error; a value that does not decode is an error naming the key.
func (r Reader) Build(ctx context.Context, module, fp string) (Key, error) {
	out := Key{Module: module, Fingerprint: fp}
	key, err := delivery.BuilderBuildKey(module, fp)
	if err != nil {
		return out, err
	}
	kv, err := r.bucket(ctx, delivery.BuilderBuildsBucket)
	if err != nil || kv == nil {
		return out, err
	}
	e, err := kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return out, nil
	}
	if err != nil {
		return out, fmt.Errorf("the builder's key %s: %w", key, err)
	}
	b, err := delivery.DecodeBuilderBuild(e.Value())
	if err != nil {
		return out, fmt.Errorf("the builder's key %s is not read: %w", key, err)
	}
	out.Found, out.Revision, out.Build = true, e.Revision(), b
	return out, nil
}

// ModuleAt is the indexer's trunk value of module written for commit (ADR 0501 §2): the newest revision of
// modules.<module> whose Commit is commit, read from the key's history; with commit empty, the value it holds now.
// None held is ErrNoValue.
func (r Reader) ModuleAt(ctx context.Context, module, commit string) (delivery.Module, error) {
	key, err := delivery.ModulesKey(module)
	if err != nil {
		return delivery.Module{}, err
	}
	kv, err := r.bucket(ctx, delivery.ModulesBucket)
	if err != nil {
		return delivery.Module{}, err
	}
	if kv == nil {
		return delivery.Module{}, fmt.Errorf("%w (%s: the indexer's bucket is not there)", ErrNoValue, module)
	}
	var values [][]byte
	if commit == "" {
		e, err := kv.Get(ctx, key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			return delivery.Module{}, fmt.Errorf("%w (%s)", ErrNoValue, module)
		}
		if err != nil {
			return delivery.Module{}, fmt.Errorf("the indexer's value of %s: %w", module, err)
		}
		values = [][]byte{e.Value()}
	} else {
		entries, err := kv.History(ctx, key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			return delivery.Module{}, fmt.Errorf("%w (%s at %s)", ErrNoValue, module, short(commit))
		}
		if err != nil {
			return delivery.Module{}, fmt.Errorf("the indexer's values of %s: %w", module, err)
		}
		for _, e := range entries {
			if e.Operation() == jetstream.KeyValuePut {
				values = append(values, e.Value())
			}
		}
	}
	return pickAt(module, commit, values)
}

// Modules is every module the indexer holds a trunk value of, by module: the value it holds now. A value that does not
// decode is left out and said in unread.
func (r Reader) Modules(ctx context.Context) (values map[string]delivery.Module, unread map[string]error, err error) {
	kv, err := r.bucket(ctx, delivery.ModulesBucket)
	if err != nil || kv == nil {
		return nil, nil, err
	}
	keys, err := kv.Keys(ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil, nil
	}
	if err != nil {
		return nil, nil, fmt.Errorf("listing the indexer's modules: %w", err)
	}
	values, unread = map[string]delivery.Module{}, map[string]error{}
	for _, key := range keys {
		e, err := kv.Get(ctx, key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			continue
		}
		if err != nil {
			return nil, nil, fmt.Errorf("the indexer's value of %s: %w", key, err)
		}
		m, err := delivery.DecodeModule(e.Value())
		if err != nil {
			unread[key] = err
			continue
		}
		values[key] = m
	}
	return values, unread, nil
}

// pickAt is, of a key's values oldest first, the newest whose commit is commit (any, with commit empty); values that
// do not decode are passed over, and one of them is said when nothing else is found.
func pickAt(module, commit string, values [][]byte) (delivery.Module, error) {
	var unread error
	for i := len(values) - 1; i >= 0; i-- {
		m, err := delivery.DecodeModule(values[i])
		if err != nil {
			if unread == nil {
				unread = err
			}
			continue
		}
		if commit == "" || sameCommit(m.Commit, commit) {
			return m, nil
		}
	}
	where := module
	if commit != "" {
		where += " at " + short(commit)
	}
	if unread != nil {
		return delivery.Module{}, fmt.Errorf("%w (%s; a value is not read: %v)", ErrNoValue, where, unread)
	}
	return delivery.Module{}, fmt.Errorf("%w (%s)", ErrNoValue, where)
}

// sameCommit is two commit ids naming one commit: equal, or one a prefix of the other of at least 7 hex digits.
func sameCommit(a, b string) bool {
	if a == b {
		return a != ""
	}
	if len(a) < 7 || len(b) < 7 {
		return false
	}
	return strings.HasPrefix(a, b) || strings.HasPrefix(b, a)
}

// Manifest is the indexer's manifest of digest: the module and its module.json byte for byte.
func (r Reader) Manifest(ctx context.Context, digest string) (delivery.Manifest, error) {
	key, err := delivery.ManifestsKey(digest)
	if err != nil {
		return delivery.Manifest{}, err
	}
	kv, err := r.bucket(ctx, delivery.ManifestsBucket)
	if err != nil {
		return delivery.Manifest{}, err
	}
	if kv == nil {
		return delivery.Manifest{}, fmt.Errorf("the manifest %s: the indexer's manifests bucket is not there", short(digest))
	}
	e, err := kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return delivery.Manifest{}, fmt.Errorf("the indexer holds no manifest %s", short(digest))
	}
	if err != nil {
		return delivery.Manifest{}, fmt.Errorf("the manifest %s: %w", short(digest), err)
	}
	m, err := delivery.DecodeManifest(e.Value())
	if err != nil {
		return delivery.Manifest{}, fmt.Errorf("the manifest %s is not read: %w", short(digest), err)
	}
	return m, nil
}

// Watch calls wake with the module of every key the builder or the indexer writes from now on, until ctx ends: what
// wakes a plan waiting on a key, with no timer (ADR 0501; the R4e context's A7). A bucket not there yet is watched once
// it is: the watch looks again each time wake would have been called for the other bucket, and at every retry.
func Watch(ctx context.Context, js jetstream.JetStream, wake func(module string)) error {
	for _, name := range []string{delivery.BuilderBuildsBucket, delivery.ModulesBucket} {
		if err := watchOne(ctx, js, name, wake); err != nil {
			return err
		}
	}
	return nil
}

func watchOne(ctx context.Context, js jetstream.JetStream, name string, wake func(module string)) error {
	kv, err := js.KeyValue(ctx, name)
	if err != nil {
		return fmt.Errorf("the bucket %s: %w", name, err)
	}
	w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
	if err != nil {
		return fmt.Errorf("watching %s: %w", name, err)
	}
	go func() {
		defer w.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case e, ok := <-w.Updates():
				if !ok {
					return
				}
				if e == nil || e.Operation() != jetstream.KeyValuePut {
					continue
				}
				module, _, _ := strings.Cut(e.Key(), ".")
				wake(module)
			}
		}
	}()
	return nil
}

func short(s string) string {
	s = strings.TrimPrefix(s, "src2:")
	if len(s) > 12 {
		return s[:12]
	}
	return s
}

func firstLine(s string) string {
	line, _, _ := strings.Cut(strings.TrimSpace(s), "\n")
	return line
}