mesh-sdk/go/stateobjects/process.go

package stateobjects

import (
	"context"
	"errors"
	"time"

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

	"git.novox.be/novox/mesh-sdk/go/kvlimit"
)

// callTimeout bounds each call a process makes of its bus.
const callTimeout = 30 * time.Second

// OverBus is a state a process reaches through its own JetStream, with the object store beside it: a foundation part's
// bucket and its old `objects` section, or a bundle-less part's `<state>-objects`. store names the object store as a
// reader asks for it, written into every reference. The bucket's own limit for one value is the limit of the values.
func OverBus(kv jetstream.KeyValue, objects jetstream.ObjectStore, store string) Values {
	limit, cancel := context.WithTimeout(context.Background(), callTimeout)
	defer cancel()
	return Values{KV: BusKV{KV: kv}, Objects: BusObjects{Store: objects}, Store: store, Limit: int(kvlimit.LimitOf(limit, kv))}
}

// BusKV is a bucket as a process reaches it.
type BusKV struct{ KV jetstream.KeyValue }

func ctx() (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.Background(), callTimeout)
}

func (b BusKV) Get(key string) (*Entry, error) {
	c, cancel := ctx()
	defer cancel()
	e, err := b.KV.Get(c, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	if e.Operation() != jetstream.KeyValuePut {
		return nil, nil
	}
	return &Entry{Value: e.Value(), Revision: e.Revision()}, nil
}

func (b BusKV) Create(key string, value []byte) error {
	c, cancel := ctx()
	defer cancel()
	_, err := kvlimit.Create(c, b.KV, key, value)
	if errors.Is(err, jetstream.ErrKeyExists) {
		return ErrChanged
	}
	return err
}

func (b BusKV) Update(key string, value []byte, revision uint64) error {
	c, cancel := ctx()
	defer cancel()
	_, err := kvlimit.Update(c, b.KV, key, value, revision)
	if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
		return ErrChanged
	}
	return err
}

func (b BusKV) Delete(key string) error {
	c, cancel := ctx()
	defer cancel()
	return b.KV.Delete(c, key)
}

func (b BusKV) Keys() ([]string, error) {
	c, cancel := ctx()
	defer cancel()
	keys, err := b.KV.Keys(c)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil
	}
	return keys, err
}

// BusObjects is an object store as a process reaches it; it takes any size in one put.
type BusObjects struct{ Store jetstream.ObjectStore }

func (BusObjects) MostBytes() int { return 0 }

func (b BusObjects) Put(name string, data []byte) error {
	c, cancel := ctx()
	defer cancel()
	_, err := b.Store.PutBytes(c, name, data)
	return err
}

func (b BusObjects) Get(name string) ([]byte, error) {
	c, cancel := ctx()
	defer cancel()
	return b.Store.GetBytes(c, name)
}

func (b BusObjects) Delete(name string) error {
	c, cancel := ctx()
	defer cancel()
	err := b.Store.Delete(c, name)
	if errors.Is(err, jetstream.ErrObjectNotFound) {
		return nil
	}
	return err
}

func (b BusObjects) List() ([]Object, error) {
	c, cancel := ctx()
	defer cancel()
	l, err := b.Store.List(c)
	if errors.Is(err, jetstream.ErrNoObjectsFound) {
		return nil, nil
	}
	out := make([]Object, 0, len(l))
	for _, o := range l {
		if !o.Deleted {
			out = append(out, Object{Name: o.Name, Size: int64(o.Size), Modified: o.ModTime})
		}
	}
	return out, err
}