Add the chain package and its buckets: builds, rollouts and assignments with history 64 (issue 1089)

internal/chain holds typed Build, Rollout and Assignment and a Backend with no delete or purge, in memory and on the bus. The builds and rollouts buckets are asserted with history 64, no age and values up to just under a mebibyte; the assignments bucket gets the same, keeping its marker age because a stream cannot be made to disallow message ages. Migration 0104 makes the one-row chain_moved switch. The controller keeps no consolidated schema, so 0104 is the whole schema change. Nothing reads the buckets yet.

internal/chain/bus.go

@@ -0,0 +1,132 @@
+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 func() bool) (*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), nil
+}
+
+// busBucket is a Backend on a key-value bucket. It calls no delete and no purge: the type has none to call.
+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) {
+	w, err := b.kv.WatchAll(ctx)
+	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 {
+		out = append(out, entryOf(e))
+	}
+	return out, nil
+}