mesh-controller/internal/conditions/bus.go

package conditions

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"sync/atomic"
	"time"

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

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

// The condition store on the bus (to-be 45 §2, ADR 0201): the controller's two buckets, asserted at
// its start like its calls and its hand-act log.

// OnTheBus opens the store and its history on a connection.
func OnTheBus(ctx context.Context, conn *nats.Conn) (Backend, History, error) {
	api, err := jetstream.New(conn)
	if err != nil {
		return nil, nil, err
	}
	open, err := api.KeyValue(ctx, broker.ConditionsBucket)
	if err != nil {
		return nil, nil, fmt.Errorf("the condition store %s is not on the bus — the controller asserts it at "+
			"its start, so one older than this has not: %w", broker.ConditionsBucket, err)
	}
	history, err := api.KeyValue(ctx, broker.ConditionHistoryBucket)
	if err != nil {
		return nil, nil, fmt.Errorf("the condition history %s is not on the bus — the controller asserts it "+
			"at its start, so one older than this has not: %w", broker.ConditionHistoryBucket, err)
	}
	return busStore{open}, &busHistory{api: api, kv: history}, nil
}

type busStore struct{ kv jetstream.KeyValue }

func (b busStore) 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 Entry{Value: e.Value(), Revision: e.Revision()}, true, nil
}

func (b busStore) Create(ctx context.Context, key string, value []byte) error {
	_, err := b.kv.Create(ctx, key, value)
	if errors.Is(err, jetstream.ErrKeyExists) {
		return ErrMoved
	}
	return err
}

func (b busStore) Update(ctx context.Context, key string, value []byte, revision uint64) error {
	_, err := b.kv.Update(ctx, key, value, revision)
	return moved(err)
}

func (b busStore) Delete(ctx context.Context, key string, revision uint64) error {
	return moved(b.kv.Delete(ctx, key, jetstream.LastRevision(revision)))
}

// moved reads the server's refusal of a compare-and-set as what it is.
func moved(err error) error {
	var apiErr *jetstream.APIError
	if errors.As(err, &apiErr) && apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence {
		return ErrMoved
	}
	return err
}

// All is every key, read through a watch that hands over each current value and then says it has.
func (b busStore) All(ctx context.Context) (map[string]Entry, error) {
	w, err := b.kv.WatchAll(ctx, jetstream.IgnoreDeletes())
	if err != nil {
		return nil, err
	}
	defer func() { _ = w.Stop() }()
	out := map[string]Entry{}
	for {
		select {
		case <-ctx.Done():
			return nil, fmt.Errorf("reading the condition store: %w", ctx.Err())
		case e := <-w.Updates():
			if e == nil {
				return out, nil
			}
			out[e.Key()] = Entry{Value: e.Value(), Revision: e.Revision()}
		}
	}
}

// busHistory keeps each transition under a key of its time and a sequence, and reads them back
// from a moment through the stream under the bucket — by time, so a read of the last ten minutes
// does not read ninety days.
type busHistory struct {
	api jetstream.JetStream
	kv  jetstream.KeyValue
	seq atomic.Uint64
}

func (h *busHistory) Append(ctx context.Context, e Event) error {
	body, err := json.Marshal(e)
	if err != nil {
		return err
	}
	key := strconv.FormatInt(e.At.UnixNano(), 10) + "-" + strconv.FormatUint(h.seq.Add(1), 10)
	_, err = h.kv.Put(ctx, key, body)
	return err
}

// historyQuiet is how long a read of the history waits for one more transition before it takes the
// stream as read to its end; it answers at once while it holds something.
const historyQuiet = 2 * time.Second

func (h *busHistory) Since(ctx context.Context, since time.Time) ([]Event, error) {
	start := since
	consumer, err := h.api.OrderedConsumer(ctx, "KV_"+broker.ConditionHistoryBucket, jetstream.OrderedConsumerConfig{
		DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &start,
	})
	if err != nil {
		return nil, fmt.Errorf("reading the condition history: %w", err)
	}
	info, err := consumer.Info(ctx)
	if err != nil {
		return nil, fmt.Errorf("reading the condition history: %w", err)
	}
	var out []Event
	pending := info.NumPending
	for pending > 0 {
		msg, err := consumer.Next(jetstream.FetchMaxWait(historyQuiet))
		if err != nil {
			if ctx.Err() != nil {
				return nil, ctx.Err()
			}
			// Nothing more within the quiet wait: read to its end.
			break
		}
		meta, err := msg.Metadata()
		if err != nil {
			break
		}
		pending = meta.NumPending
		var e Event
		if len(msg.Data()) > 0 && json.Unmarshal(msg.Data(), &e) == nil && e.Key != "" {
			out = append(out, e)
		}
	}
	sort.SliceStable(out, func(i, j int) bool { return out[i].At.Before(out[j].At) })
	return out, nil
}