mesh-controller/internal/chain/bus.go

package chain

import (
	"context"
	"errors"
	"fmt"

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

	"github.com/novox/mesh-controller/internal/broker"
)

// OnTheBus opens the chain's three buckets on a connection. The controller asserts them at its start
// (broker.EnsureControllerBuckets), so one older than this has not, and the error says so.
func OnTheBus(ctx context.Context, conn *nats.Conn, holds Holds) (*Chain, error) {
	api, err := jetstream.New(conn)
	if err != nil {
		return nil, err
	}
	open := func(bucket string) (Backend, error) {
		kv, err := api.KeyValue(ctx, bucket)
		if err != nil {
			return nil, fmt.Errorf("the bucket %s is not on the bus — the controller asserts it at its start, so one "+
				"older than this has not: %w", bucket, err)
		}
		return busBucket{kv}, nil
	}
	builds, err := open(broker.BuildsBucket)
	if err != nil {
		return nil, err
	}
	rollouts, err := open(broker.RolloutsBucket)
	if err != nil {
		return nil, err
	}
	assignments, err := open(broker.AssignmentsBucket)
	if err != nil {
		return nil, err
	}
	return New(builds, rollouts, assignments, holds).WithManifests(builds), nil
}

// busBucket is a Backend on a key-value bucket. It calls no delete and no purge: the type has none to call (ADR 0408 §4).
type busBucket struct{ kv jetstream.KeyValue }

func entryOf(e jetstream.KeyValueEntry) Entry {
	return Entry{Key: e.Key(), Value: e.Value(), Revision: e.Revision(), At: e.Created()}
}

func (b busBucket) Get(ctx context.Context, key string) (Entry, bool, error) {
	e, err := b.kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return Entry{}, false, nil
	}
	if err != nil {
		return Entry{}, false, err
	}
	return entryOf(e), true, nil
}

func (b busBucket) Create(ctx context.Context, key string, value []byte) (uint64, error) {
	rev, err := b.kv.Create(ctx, key, value)
	if errors.Is(err, jetstream.ErrKeyExists) {
		return 0, ErrExists
	}
	return rev, err
}

func (b busBucket) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) {
	rev, err := b.kv.Update(ctx, key, value, revision)
	var apiErr *jetstream.APIError
	if errors.As(err, &apiErr) && apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence {
		return 0, ErrMoved
	}
	return rev, err
}

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

func (b busBucket) Watch(ctx context.Context) (<-chan Entry, error) {
	return b.WatchFiltered(ctx, ">")
}

func (b busBucket) WatchFiltered(ctx context.Context, filter string) (<-chan Entry, error) {
	var w jetstream.KeyWatcher
	var err error
	if filter == ">" {
		w, err = b.kv.WatchAll(ctx, jetstream.IgnoreDeletes())
	} else {
		w, err = b.kv.WatchFiltered(ctx, []string{filter}, jetstream.IgnoreDeletes())
	}
	if err != nil {
		return nil, err
	}
	out := make(chan Entry)
	go func() {
		defer close(out)
		defer func() { _ = w.Stop() }()
		for {
			select {
			case <-ctx.Done():
				return
			case e, ok := <-w.Updates():
				if !ok {
					return
				}
				v := Entry{Caught: true}
				if e != nil {
					v = entryOf(e)
				}
				select {
				case out <- v:
				case <-ctx.Done():
					return
				}
			}
		}
	}()
	return out, nil
}

func (b busBucket) History(ctx context.Context, key string) ([]Entry, error) {
	all, err := b.kv.History(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	out := make([]Entry, 0, len(all))
	for _, e := range all {
		if e.Operation() != jetstream.KeyValuePut {
			continue // a marker an earlier mirror left; nothing of the chain makes one
		}
		out = append(out, entryOf(e))
	}
	return out, nil
}