mesh-sdk/go/state.go

package stdio

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

// State — the current value of something a module keeps on the bus, which every machine sees,
// including one that joins later (novox/hq ADR 0201). Not history: that is an event. Never a secret,
// sealed or not. A module declares what it keeps (`state`) and what it reads (`reads`, as
// `<module>.<name>`) and names each here as it named it there; the runtime refuses, with the reason,
// what it did not declare.

// StateEntry is one key's current value.
type StateEntry struct {
	Key      string          `json:"key"`
	Value    json.RawMessage `json:"value"`
	Revision uint64          `json:"revision"`
}

// StateChange is one change a watch hands over: a key put or deleted.
type StateChange struct {
	State    string          `json:"state"`
	Key      string          `json:"key"`
	Op       string          `json:"op"`
	Value    json.RawMessage `json:"value,omitempty"`
	Revision uint64          `json:"revision"`
	// Current is true for a value that was there when the watch began, false for a change since.
	Current bool `json:"current"`
}

// StateRevision is one value a bucket still keeps for a key: what was written, at which revision, by which
// operation ("put" or "delete", whose value is null) and when (novox/hq ADR 0307).
type StateRevision struct {
	Key      string          `json:"key"`
	Value    json.RawMessage `json:"value"`
	Revision uint64          `json:"revision"`
	Op       string          `json:"op"`
	At       time.Time       `json:"at"`
}

// KeptState is one state this module keeps, or reads.
type KeptState struct{ name string }

// State is a state this module keeps (by its local name) or reads (`<module>.<name>`).
func State(name string) KeptState { return KeptState{name: name} }

const stateTimeout = 30 * time.Second

func (s KeptState) ask(verb string, params map[string]any) (json.RawMessage, error) {
	c, err := live()
	if err != nil {
		return nil, err
	}
	params["state"] = s.name
	return c.ask("mesh/state."+verb, params, stateTimeout)
}

// Get is a key's current value, or nil when it has none.
func (s KeptState) Get(key string) (*StateEntry, error) {
	raw, err := s.ask("get", map[string]any{"key": key})
	if err != nil || len(raw) == 0 || string(raw) == "null" {
		return nil, err
	}
	var e StateEntry
	return &e, json.Unmarshal(raw, &e)
}

// History is every value the bucket still keeps for key, oldest first — as many as the state declared it
// keeps, deletes included — and empty for a key it holds nothing for (novox/hq ADR 0307). A read, like Get:
// the owner's instances and its readers may ask it, and it writes nothing.
func (s KeptState) History(key string) ([]StateRevision, error) {
	raw, err := s.ask("history", map[string]any{"key": key})
	if err != nil {
		return nil, err
	}
	revisions := []StateRevision{}
	if len(raw) == 0 || string(raw) == "null" {
		return revisions, nil
	}
	return revisions, json.Unmarshal(raw, &revisions)
}

// Put writes a key — the owner's instances only — and returns the revision written.
func (s KeptState) Put(key string, value any) (uint64, error) {
	raw, err := s.ask("put", map[string]any{"key": key, "value": value})
	if err != nil {
		return 0, err
	}
	var put struct {
		Revision uint64 `json:"revision"`
	}
	return put.Revision, json.Unmarshal(raw, &put)
}

// Delete removes a key — the owner's instances only. A key that was not there is not an error.
func (s KeptState) Delete(key string) error {
	_, err := s.ask("delete", map[string]any{"key": key})
	return err
}

// Increment adds one to a counter kept at key and returns the number it now holds: one for a key with no
// value. Read, then written only if the key is still at the revision read (Create when it had none, Update
// when it had one), so of any number of writers each number is handed out once and no increment is lost.
// A write another writer won is read again and tried again, at most attempts times in all; past that the
// error is ErrChanged and nothing was counted. Any other error is returned at once, as is a value that is
// not a whole number (null included): nothing is ever reported counted that was not written. The reverse
// does not hold: an error that is not ErrChanged — a timeout above all — may come after the write landed, so
// that number may be spent and never handed out. A caller tolerates gaps, never a number handed out twice.
// Deleting the key starts the count again at one; a key whose numbers must stay unique is never deleted.
func (s KeptState) Increment(key string, attempts int) (uint64, error) {
	if attempts < 1 {
		return 0, fmt.Errorf("an increment of %s is tried at least once: %d attempts", key, attempts)
	}
	for i := 0; i < attempts; i++ {
		e, err := s.Get(key)
		if err != nil {
			return 0, err
		}
		var next uint64 = 1
		if e == nil {
			_, err = s.Create(key, next)
		} else {
			var n uint64
			if string(e.Value) == "null" {
				return 0, fmt.Errorf("%s holds null, not a counter", key)
			}
			if err := json.Unmarshal(e.Value, &n); err != nil {
				return 0, fmt.Errorf("%s holds %s, not a counter: %w", key, e.Value, err)
			}
			if n == ^uint64(0) {
				return 0, fmt.Errorf("%s is at the largest number a counter holds", key)
			}
			next = n + 1
			_, err = s.Update(key, next, e.Revision)
		}
		if err == nil {
			return next, nil
		}
		if !errors.Is(err, ErrChanged) {
			return 0, err
		}
	}
	return 0, fmt.Errorf("%s was written by another writer first %d times: %w", key, attempts, ErrChanged)
}

// Keys is every key with a current value, sorted.
func (s KeptState) Keys() ([]string, error) {
	raw, err := s.ask("keys", map[string]any{})
	if err != nil {
		return nil, err
	}
	var keys []string
	return keys, json.Unmarshal(raw, &keys)
}

// Watch hands handle every current value — none that is deleted — 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 watch lasts as long as the process.
func (s KeptState) Watch(key string, handle func(StateChange) error) error {
	c, err := live()
	if err != nil {
		return err
	}
	id := c.newID()
	c.mu.Lock()
	c.watches[id] = handle
	c.mu.Unlock()
	if _, err := c.askAs(id, "mesh/state.watch", map[string]any{"state": s.name, "key": key}, stateTimeout); err != nil {
		c.mu.Lock()
		delete(c.watches, id)
		c.mu.Unlock()
		return err
	}
	return nil
}