mesh-sdk/go/process/state.go

package process

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

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

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

// stateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a denied publish is
// caught at once by Conn's own ErrorHandler (conn.go withRefusal/publishRefused), never waited out, so
// this stays the ordinary request bound and is not exposed for a test to shrink.
const stateTimeout = 30 * time.Second

// State is one of this process's module's states, read and written on this connection (novox/hq ADR 0436
// clause 4), in the bucket <module>_<name> (mesh-controller internal/broker/state.go:49) — only as the
// account's grants allow. It satisfies go/asker's Store.
type State struct {
	Module string
	Name   string
	// kv binds this state's underlying key-value store: a live connection's own (Conn.State), or, in a
	// unit test, a fake that never reaches a bus.
	kv func(ctx context.Context) (jetstream.KeyValue, error)
}

// bucket is this state's name on the bus.
func (s State) bucket() string { return s.Module + "_" + s.Name }

// State is the named state this connection's module keeps, bound lazily to its bucket on each call so a
// bucket the account is not granted fails at the call that needed it, naming the bucket. The bind itself
// (js.KeyValue) is what asks $JS.API.STREAM.INFO.KV_<bucket>, so it is what withRefusal watches: a bucket
// not granted is refused there, promptly, as nats.ErrPermissionViolation.
func (c *Conn) State(name string) State {
	bucket := c.account.Module + "_" + name
	return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
		if err := c.refusedUnannounced(name); err != nil {
			return nil, err
		}
		js, err := jetstream.New(c.nc)
		if err != nil {
			return nil, err
		}
		subject := "$JS.API.STREAM.INFO.KV_" + bucket
		ctx, stop := c.withRefusal(ctx, subject)
		defer stop()
		kv, err := js.KeyValue(ctx, bucket)
		if err != nil {
			if errors.Is(context.Cause(ctx), errPermissionRefused) {
				return nil, fmt.Errorf("%w: publishing %s", nats.ErrPermissionViolation, subject)
			}
			return nil, err
		}
		return kv, nil
	}}
}

func (s State) store(ctx context.Context) (jetstream.KeyValue, error) {
	kv, err := s.kv(ctx)
	if err != nil {
		return nil, fmt.Errorf("%s's state %s (bucket %s): %w", s.Module, s.Name, s.bucket(), err)
	}
	return kv, nil
}

// Get is a key's current value, or nil when it has none.
func (s State) Get(key string) (*stdio.StateEntry, error) {
	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
	defer cancel()
	kv, err := s.store(ctx)
	if err != nil {
		return nil, err
	}
	entry, err := kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
	}
	return &stdio.StateEntry{Key: key, Value: json.RawMessage(entry.Value()), Revision: entry.Revision()}, nil
}

// Create writes a key only when it has none, and returns the revision; a key that has one is refused as
// stdio.ErrChanged.
func (s State) Create(key string, value any) (uint64, error) {
	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
	defer cancel()
	kv, err := s.store(ctx)
	if err != nil {
		return 0, err
	}
	data, err := json.Marshal(value)
	if err != nil {
		return 0, err
	}
	rev, err := kv.Create(ctx, key, data)
	if errors.Is(err, jetstream.ErrKeyExists) {
		return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
	}
	if err != nil {
		return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
	}
	return rev, nil
}

// Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile
// is refused as stdio.ErrChanged.
func (s State) Update(key string, value any, revision uint64) (uint64, error) {
	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
	defer cancel()
	kv, err := s.store(ctx)
	if err != nil {
		return 0, err
	}
	data, err := json.Marshal(value)
	if err != nil {
		return 0, err
	}
	rev, err := kv.Update(ctx, key, data, revision)
	if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
		return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
	}
	if err != nil {
		return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
	}
	return rev, nil
}