123 lines
3.2 KiB
Go
123 lines
3.2 KiB
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)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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.ParseManifestForInputs([]byte(kept.JSON))
|
|
}
|