mesh-sdk/go/process/state.go

package process

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

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

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

// 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 operator-channel's client's Store (mesh-contracts operatorchannelclient).
type State struct {
	Module string
	Name   string
	// bucketName is the bucket when it is not <module>_<name>: one node's bucket of a per-node state (NodeFamily).
	bucketName 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)
	// closed says the connection is gone, so a watch ends; nil in a unit test.
	closed func() bool
}

// bucket is this state's name on the bus.
func (s State) bucket() string {
	if s.bucketName != "" {
		return s.bucketName
	}
	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 {
	return c.bound(c.account.Module, name, "", func() error { return c.refusedUnannounced(name) })
}

// Read is another module's state this connection's module reads, named as its manifest's `reads` names it
// (`<module>.<state>`, novox/hq ADR 0495): bound like State to the bucket <module>_<state>, so Get, Keys and Watch
// work as far as the account's read grant goes, and a write is refused by the bus (issue 774: route-proxy follows
// the seat mesh-provisioner's plans). A name that is not `<module>.<state>` fails at the call that needed it.
func (c *Conn) Read(read string) State {
	module, name, ok := strings.Cut(read, ".")
	return c.bound(module, name, "", func() error {
		if !ok || module == "" || name == "" || strings.Contains(name, ".") {
			return fmt.Errorf("%q is not a state another module keeps, <module>.<state>", read)
		}
		return nil
	})
}

// bound is module's state name on this connection, in its bucket <module>_<name> or the bucket given, checked by
// refused before every bind.
func (c *Conn) bound(module, name, bucket string, refused func() error) State {
	given := bucket
	if bucket == "" {
		bucket = module + "_" + name
	}
	return State{Module: module, Name: name, bucketName: given, closed: func() bool { return c.nc.IsClosed() }, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
		if err := refused(); 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.
// Get waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
func (s State) Get(key string) (*stdio.StateEntry, error) {
	return busretry.Do(s.what("get "+key), func() (*stdio.StateEntry, error) { return s.get(key) }, nil)
}

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.
// Create waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
func (s State) Create(key string, value any) (uint64, error) {
	return busretry.Do(s.what("create "+key), func() (uint64, error) { return s.create(key, value) }, nil)
}

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 := kvlimit.Create(ctx, kv, 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
}

// MinTTL is the shortest time to live a key may be given: the bus counts a key's age in whole seconds.
const MinTTL = time.Second

// CreateTTL writes a key only when it has none, with its own time to live (novox/hq ADR 0564 clauses 4 and 7, issue
// 1952 S6): the bus removes it when ttl passes and its watchers hear the removal, so an expiry is an event and no timer
// runs. A key that has a value is refused as stdio.ErrChanged: an act begun once cannot be begun again. The bucket must
// take keys with their own time to live (a state declared `key-ttl`).
// CreateTTL waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
func (s State) CreateTTL(key string, value any, ttl time.Duration) (uint64, error) {
	return busretry.Do(s.what("create "+key), func() (uint64, error) { return s.createTTL(key, value, ttl) }, nil)
}

func (s State) createTTL(key string, value any, ttl time.Duration) (uint64, error) {
	if ttl < MinTTL {
		return 0, fmt.Errorf("%s's state %s, key %s: a time to live of %s is shorter than the bus keeps (%s)", s.Module, s.Name, key, ttl, MinTTL)
	}
	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 := kvlimit.Create(ctx, kv, key, data, jetstream.KeyTTL(ttl))
	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
}

// RenewTTL writes a key whole with a fresh time to live, whatever it held: a pulse its signal renews (ADR 0564 clause
// 4). A time to live is set only when a key is created, so a live key is deleted first; its watchers hear the delete and
// then the new value.
func (s State) RenewTTL(key string, value any, ttl time.Duration) (uint64, error) {
	rev, err := s.CreateTTL(key, value, ttl)
	if !errors.Is(err, stdio.ErrChanged) {
		return rev, err
	}
	if err := s.Delete(key); err != nil {
		return 0, err
	}
	return s.CreateTTL(key, value, ttl)
}

// 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.
// Update waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
func (s State) Update(key string, value any, revision uint64) (uint64, error) {
	return busretry.Do(s.what("update "+key), func() (uint64, error) { return s.update(key, value, revision) }, nil)
}

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 := kvlimit.Update(ctx, kv, 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
}

// Put writes a key whole, whatever it held, and returns the revision written.
// Put waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
func (s State) Put(key string, value any) (uint64, error) {
	return busretry.Do(s.what("put "+key), func() (uint64, error) { return s.put(key, value) }, nil)
}

func (s State) put(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 := kvlimit.Put(ctx, kv, key, data)
	if err != nil {
		return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
	}
	return rev, nil
}

// Keys is every key with a current value, sorted; none is an empty list.
// Keys waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
func (s State) Keys() ([]string, error) {
	return busretry.Do(s.what("keys"), func() ([]string, error) { return s.keys() }, nil)
}

func (s State) keys() ([]string, error) {
	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
	defer cancel()
	kv, err := s.store(ctx)
	if err != nil {
		return nil, err
	}
	keys, err := kv.Keys(ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return []string{}, nil
	}
	if err != nil {
		return nil, fmt.Errorf("%s's state %s, keys: %w", s.Module, s.Name, err)
	}
	sort.Strings(keys)
	return keys, nil
}

// watchPattern is a key narrowing as the bus names it: `*` stays one name, `**` is the rest, empty is every key.
func watchPattern(key string) string {
	if key == "" {
		return jetstream.AllKeys
	}
	tokens := strings.Split(key, ".")
	for i, t := range tokens {
		if t == "**" {
			tokens[i] = ">"
		}
	}
	return strings.Join(tokens, ".")
}

// Watch hands handle every current value — none that is deleted, each with Current set — and then every
// change in order, and returns once the current values have all been handled. key narrows it: a key, with
// `*` for one name and `**` for the rest; empty is every key. The changes go on for the life of the
// connection; an error from handle, or the connection closing, ends them. Before the current values are
// handled, either is returned as the error.
func (s State) Watch(key string, handle func(stdio.StateChange) error) error {
	// the watch outlives this call, so its context is not the call's timeout: the connection ends it.
	ctx, cancel := context.WithCancel(context.Background())
	bind, stopBind := context.WithTimeout(ctx, stateTimeout)
	kv, err := s.store(bind)
	stopBind()
	if err != nil {
		cancel()
		return err
	}
	w, err := kv.Watch(ctx, watchPattern(key))
	if err != nil {
		cancel()
		return fmt.Errorf("%s's state %s, watching %q: %w", s.Module, s.Name, key, err)
	}
	end := func() { w.Stop(); cancel() }
	next := func() (jetstream.KeyValueEntry, bool, error) {
		tick := time.NewTicker(250 * time.Millisecond)
		defer tick.Stop()
		for {
			select {
			case e, open := <-w.Updates():
				if !open {
					return nil, false, fmt.Errorf("%s's state %s: the watch ended", s.Module, s.Name)
				}
				return e, true, nil
			case <-tick.C:
				if s.closed != nil && s.closed() {
					return nil, false, fmt.Errorf("%s's state %s: the connection closed", s.Module, s.Name)
				}
			}
		}
	}
	deliver := func(e jetstream.KeyValueEntry, current bool) error {
		c := stdio.StateChange{State: s.Name, Key: e.Key(), Op: "put", Revision: e.Revision(), Current: current}
		if e.Operation() == jetstream.KeyValuePut {
			c.Value = json.RawMessage(e.Value())
		} else {
			c.Op = "delete"
		}
		return handle(c)
	}
	for {
		e, _, err := next()
		if err != nil {
			end()
			return err
		}
		if e == nil {
			break
		}
		if e.Operation() != jetstream.KeyValuePut {
			continue
		}
		if err := deliver(e, true); err != nil {
			end()
			return err
		}
	}
	go func() {
		defer end()
		for {
			e, _, err := next()
			if err != nil || e == nil {
				return
			}
			if deliver(e, false) != nil {
				return
			}
		}
	}()
	return nil
}

// Delete removes a key. A key that was not there is not an error.
// Delete waits out a bus that answers nothing for a few seconds, as Put does.
func (s State) Delete(key string) error {
	_, err := busretry.Do(s.what("delete "+key), func() (struct{}, error) { return struct{}{}, s.delete(key) }, nil)
	return err
}

// what names a request on this state for busretry.
func (s State) what(request string) string {
	return fmt.Sprintf("%s's state %s, %s", s.Module, s.Name, request)
}

func (s State) delete(key string) error {
	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
	defer cancel()
	kv, err := s.store(ctx)
	if err != nil {
		return err
	}
	if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
		return fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
	}
	return nil
}