mesh-sdk/go/catalogue/busstate.go

package catalogue

import (
	"bytes"
	"encoding/json"
	"fmt"
	"slices"
	"sort"
	"strings"
)

// Bus state is what a module keeps on the bus, asked for as a provision of the bus module in `requires` (novox/hq ADR
// 0595 §1, design 32 § 4): an entry of `requires` in its typed long form, the local name, the storage type, the template
// and the template's options. This is the contract only (ADR 0578 §7, ADR 0593 §1): the entry's shape, decoded
// leniently. Every rule that judges an entry (a template known, an option within its bound, a fixed setting not given)
// is the indexer's judge's.

// The storage types (design 25 § 12.1): the stream is the base; the key-value bucket, the object store and the ledger
// are built on it, each with calls of its own (ADR 0595 §2).
const (
	StorageStream      = "stream"
	StorageKeyValue    = "key-value"
	StorageObjectStore = "object-store"
	StorageLedger      = "ledger"
)

// The six templates, each a storage type with a predefined configuration (ADR 0595 §2, design 25 § 12.1): state on
// key-value, events, work queue and log on stream, objects on object store, ledger on ledger. A bare stream names
// none and gives its own Configuration.
const (
	TemplateState     = "state"
	TemplateEvents    = "events"
	TemplateWorkQueue = "work-queue"
	TemplateLog       = "log"
	TemplateObjects   = "objects"
	TemplateLedger    = "ledger"
)

// The option names, a template's settings a module may change (design 25 § 12.1's table), written as the old sections
// wrote them, so a module that moves carries its old settings as options unchanged (ADR 0595 §11). Which option a
// template takes, and its bound, is the judge's.
const (
	// OptionPastValues is how many past values a state's key keeps (the old `state`'s `history`).
	OptionPastValues = "history"
	// OptionTimeToLive is how long a state's value lives, in seconds.
	OptionTimeToLive = "ttl-seconds"
	// OptionLifetimePerKey is whether a state's key may be given a lifetime of its own.
	OptionLifetimePerKey = "key-ttl"
	// OptionPerMachine is whether a state is kept one bucket per machine.
	OptionPerMachine = "per-machine"
	// OptionAgeDays is how long a stream keeps a message, in days.
	OptionAgeDays = "max-age-days"
	// OptionCapMiB is the cap of a stream, a bucket or an object store, in MiB.
	OptionCapMiB = "max-mib"
	// OptionGrant names the trusted config value granting another module a part of it (ADR 0363 §4, ADR 0592 §9).
	OptionGrant = "grant"
	// OptionDeliveries is how many times a work queue's message is delivered before it is a dead letter.
	OptionDeliveries = "max-deliveries"
	// OptionDuplicates is how long, in seconds, a work queue detects a message sent twice.
	OptionDuplicates = "duplicates"
)

// ConfigWhenFull is what a configuration of its own says a bare stream does when full: WhenFullRefuse or
// WhenFullDropOldest. In a template it is fixed, never an option (ADR 0595 §3).
const (
	ConfigWhenFull     = "when-full"
	WhenFullRefuse     = "refuse"
	WhenFullDropOldest = "drop-oldest"
)

// The ledger effects, what an operation does to an instance's state (ADR 0597 §2): one of this fixed set.
const (
	// EffectSetFields sets the state's fields to the operation's Fields.
	EffectSetFields = "set-fields"
	// EffectSetEntry sets one entry of the state's map Into, under the key the operation's field Key holds, to its Fields.
	EffectSetEntry = "set-entry"
	// EffectAppend appends the operation's Fields to the state's list Into.
	EffectAppend = "append"
	// EffectRemove removes the entry of the map Into under the operation's field Key, or, naming no Into, the instance's
	// state.
	EffectRemove = "remove"
)

// LedgerType is one type of a ledger (ADR 0597 §1): a schema and the operations allowed on it, with its indexes. An
// instance of it is named by a path whose first part is the type's name (`decision/0290`).
type LedgerType struct {
	Name string `json:"name"`
	// Schema is the type's fields, by name.
	Schema map[string]LedgerField `json:"schema"`
	// Operations are the operations allowed on an instance, each with its effect.
	Operations []LedgerOperation `json:"operations"`
	// Indexes are its declared indexes, each a second key-value bucket kept by a watcher (ADR 0597 §4).
	Indexes []LedgerIndex `json:"indexes,omitempty"`
}

// LedgerField is one field of a ledger type's schema.
type LedgerField struct {
	// Type is the field's value type, as the schema names it.
	Type string `json:"type"`
	// DigestOf names the text field this field is the digest of (ADR 0597 §2): `apply` refuses an operation whose
	// digest does not match its text.
	DigestOf string `json:"digest-of,omitempty"`
}

// LedgerOperation is one operation a ledger type allows, and its effect on the state.
type LedgerOperation struct {
	Name string `json:"name"`
	// Effect is one of the ledger effects (EffectSetFields, EffectSetEntry, EffectAppend, EffectRemove).
	Effect string `json:"effect"`
	// Fields are the operation's fields its effect copies into the state; it may carry more (ADR 0597 §2).
	Fields []string `json:"fields,omitempty"`
	// Into is the state's map (EffectSetEntry, EffectRemove) or list (EffectAppend) the effect writes.
	Into string `json:"into,omitempty"`
	// Key is the operation's field holding the entry's key (EffectSetEntry, EffectRemove).
	Key string `json:"key,omitempty"`
	// Sequence and Author are whether the effect also writes the operation's sequence and its proven author.
	Sequence bool `json:"sequence,omitempty"`
	Author   bool `json:"author,omitempty"`
}

// LedgerIndex is one declared index of a ledger type: its name, and the fields it is kept on.
type LedgerIndex struct {
	Name string   `json:"name"`
	On   []string `json:"on"`
}

// BusStateField is where a stored manifest keeps its bus-state entries: beside `requires`, which stays a plain list of
// names, as OptionalRequiresKey keeps the optional ones.
const BusStateField = "bus-state"

// BusStateEntry is one piece of a module's bus state, as its manifest asks for it.
type BusStateEntry struct {
	// Name is its local name; the mesh, never the module, names it on the bus (ADR 0201 §4).
	Name string `json:"name"`
	// Storage is its storage type (design 25 § 12.1): how the bus keeps it.
	Storage string `json:"storage"`
	// Template is the predefined configuration of that storage type it uses; empty for a bare stream, which gives
	// Configuration instead.
	Template string `json:"template,omitempty"`
	// Options are the template's settings the module changes, by option name; each value as the manifest wrote it.
	Options map[string]json.RawMessage `json:"options,omitempty"`
	// Configuration is a configuration of the module's own, where no template fits: a bare stream's, or a ledger's
	// whose history expires (ADR 0595 §2).
	Configuration map[string]json.RawMessage `json:"configuration,omitempty"`
	// Types are a ledger's types (ADR 0597 §1), for the storage type ledger only.
	Types []LedgerType `json:"types,omitempty"`
}

// isBusStateEntry is whether a `requires` item in its long form asks for bus state: it names a storage type.
func isBusStateEntry(raw json.RawMessage) bool {
	var keys map[string]json.RawMessage
	if json.Unmarshal(raw, &keys) != nil {
		return false
	}
	_, has := keys["storage"]
	return has
}

// readBusStateEntry reads one bus-state entry leniently (ADR 0578 §3): a key it does not know, at any depth, is set
// aside and its path named, never refused. An error is only an entry of the wrong shape.
func readBusStateEntry(raw json.RawMessage) (BusStateEntry, []string, error) {
	var tree any
	dec := json.NewDecoder(bytes.NewReader(raw))
	dec.UseNumber()
	if err := dec.Decode(&tree); err != nil {
		return BusStateEntry{}, nil, err
	}
	var aside []string
	strict := func(e *BusStateEntry) error {
		b, err := json.Marshal(tree)
		if err != nil {
			return err
		}
		d := json.NewDecoder(bytes.NewReader(b))
		d.DisallowUnknownFields()
		return typedUnknown(d.Decode(e))
	}
	for tries := 0; ; tries++ {
		var e BusStateEntry
		err := strict(&e)
		if err == nil {
			return e, aside, nil
		}
		unknown := asUnknownField(err)
		if unknown == nil || tries > 64 {
			return BusStateEntry{}, nil, err
		}
		path, ok := removalThatHelps(tree, unknown.Field, err.Error(), func() error { var e BusStateEntry; return strict(&e) })
		if !ok {
			if removeEvery(tree, unknown.Field) == 0 {
				return BusStateEntry{}, nil, err
			}
			path = "…." + unknown.Field
		}
		aside = append(aside, path)
	}
}

// setAsideAlso adds paths read without to what a manifest set aside (SetAside), in its words.
func setAsideAlso(unknown string, paths []string) string {
	if len(paths) == 0 {
		return unknown
	}
	if unknown == "" {
		field := paths[0][strings.LastIndex(paths[0], ".")+1:]
		unknown = fmt.Sprintf("json: unknown field %q", field)
	}
	return unknown + " (read without " + strings.Join(paths, ", ") + ")"
}

// KeptStates are the local names of the state a manifest keeps, each a key-value bucket of its own: the old `state`
// section's and every key-value entry it asks for in `requires` (ADR 0595 §1, §11), sorted, once each. It is the one
// rule every reader of what a module keeps calls (novox/hq issue 1966): a process's `bus.state` and a seat
// contribution's state name one of these, in either form.
func KeptStates(m Manifest) []string {
	var out []string
	for _, s := range m.State {
		out = append(out, s.Name)
	}
	for _, e := range m.BusState {
		if e.Storage == StorageKeyValue {
			out = append(out, e.Name)
		}
	}
	sort.Strings(out)
	return slices.Compact(out)
}