mesh-tools/node-tools/internal/bus/state.go

package bus

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

	"git.novox.be/novox/mesh-sdk/go/membership"
	"git.novox.be/novox/mesh-sdk/go/pernode"
	"github.com/nats-io/nats.go/jetstream"
)

// A module's state on the bus (novox/hq ADR 0201): key-value buckets the mesh creates from what
// the module declared, and issues to each assignment in its membership by the name the module uses —
// its own state by the local name, another module's as `<module>.<name>`.
//
// **The runtime keeps each module to its own buckets.** One account per machine carries every module
// on it, so the bus enforces only the union; and a write the bus refuses reaches the writer as a
// timeout, not a refusal (measured, novox/hq research 024). So what a module may reach is decided here,
// from its membership, and refused with the reason before anything is sent.

// StateIssued is one bucket an assignment may reach (ADR 0201).
type StateIssued struct {
	Name   string `json:"name"`
	Bucket string `json:"bucket"`
	Writes bool   `json:"writes,omitempty"`
	// PerNode says the state is kept one bucket per node (novox/hq ADR 0593 §6, mesh-sdk membership.StateIssued): Bucket is
	// then the buckets' prefix, each bucket the prefix and a node's name, and a key's first part names its node. It is read
	// only, served as one state (per_node.go).
	PerNode bool `json:"per-node,omitempty"`
}

// StateEntry is one key's current value.
type StateEntry struct {
	Key      string          `json:"key"`
	Value    json.RawMessage `json:"value"`
	Revision uint64          `json:"revision"`
	// At is when the bus wrote this value (the entry's created time), as stdio.StateEntry.At reads it.
	At time.Time `json:"at,omitzero"`
}

// StateChange is one change a watch delivers: a key put, deleted, or expired (its own time to live, or the
// bucket's age, ended; issue 1952).
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"`
}

// StateTimeout bounds one state operation on the bus.
var StateTimeout = 10 * time.Second

// stateKey is a key the bus can hold: letters, digits and `-/_=.`, no leading or trailing dot. A
// module's convention for naming a machine in a key (`all.<server>`, `<machine>.<server>`) fits it.
var stateKey = regexp.MustCompile(`^[-/_=a-zA-Z0-9]+(\.[-/_=a-zA-Z0-9]+)*$`)

// issuedState is the bucket a module may reach by a name, and whether it may write it — or the reason
// it may not reach it at all.
func (c *Conn) issuedState(module, name string) (StateIssued, error) {
	m := c.Membership(module)
	if m == nil {
		return StateIssued{}, fmt.Errorf("%s has no membership issued on %s yet, so no state of it is reachable "+
			"until the mesh issues one (novox/hq ADR 0201)", module, c.node)
	}
	// What a membership issues by a local name is mesh-sdk's one rule (membership.StateIssued, issue 1966): the old
	// `state` list, then a key-value entry the module asks for in `requires` (ADR 0595 §1), by its local name.
	issuing := membership.Membership{Bindings: m.Bindings}
	for _, s := range m.State {
		issuing.State = append(issuing.State, membership.StateIssued{Name: s.Name, Bucket: s.Bucket, Writes: s.Writes, PerNode: s.PerNode})
	}
	if s, ok := issuing.StateIssued(name); ok {
		return StateIssued{Name: s.Name, Bucket: s.Bucket, Writes: s.Writes, PerNode: s.PerNode}, nil
	}
	issued := "none"
	if names := issuing.StateNames(); len(names) > 0 {
		issued = strings.Join(names, ", ")
	}
	return StateIssued{}, fmt.Errorf("%s keeps and reads no state called %q: it declares what it keeps under "+
		"`state` or as a key-value entry in `requires`, and what it reads under `reads` as <module>.<name>, and was "+
		"issued: %s (novox/hq ADR 0201, ADR 0595 §1)", module, name, issued)
}

func (c *Conn) bucket(ctx context.Context, s StateIssued) (jetstream.KeyValue, error) {
	js, err := jetstream.New(c.nc)
	if err != nil {
		return nil, err
	}
	kv, err := js.KeyValue(ctx, s.Bucket)
	if err != nil {
		if errors.Is(err, jetstream.ErrBucketNotFound) {
			return nil, fmt.Errorf("the state %q is issued and its bucket is not on the bus yet: the mesh "+
				"creates it from the catalogue on its next raise", s.Name)
		}
		return nil, err
	}
	return kv, nil
}

func checkKey(key string) error {
	if !stateKey.MatchString(key) {
		return fmt.Errorf("%q is not a key the bus can hold: letters, digits and -/_=, in dot-separated "+
			"names", key)
	}
	return nil
}

// StateGet is one key's current value, or nil when it has none.
func (c *Conn) StateGet(module, name, key string) (*StateEntry, error) {
	s, err := c.issuedState(module, name)
	if err != nil {
		return nil, err
	}
	if err := checkKey(key); err != nil {
		return nil, err
	}
	if s.PerNode {
		return c.perNodeGet(s, key)
	}
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	kv, err := c.bucket(ctx, s)
	if err != nil {
		return nil, err
	}
	e, err := kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision(), At: e.Created().UTC()}, nil
}

// StatePage is the current values after a key, in key order, and the key to ask after for the rest.
type StatePage struct {
	Entries []StateEntry `json:"entries"`
	Next    string       `json:"next,omitempty"`
}

// StatePageBudget is the time a page spends reading when the asker names none: well under the 30 s a reader waits.
const StatePageBudget = 10 * time.Second

// PageClock is the time a page's budget counts by, replaced by a test.
var PageClock = time.Now

// StatePage reads the values in key order after `after` in one pass, as many as fit maxBytes (each key and value),
// stopping when budget is spent, at least one value, so a bundle asks one request for many values, where StateGet
// is one request each. Access is checked once and the bucket bound once. It is a read like StateGet: served to the
// owner and to every reader, and writes nothing. Next is the last key given when more follow.
func (c *Conn) StatePage(module, name, after string, maxBytes int, budget time.Duration) (StatePage, error) {
	s, err := c.issuedState(module, name)
	if err != nil {
		return StatePage{}, err
	}
	if budget <= 0 {
		budget = StatePageBudget
	}
	// The bind and the listing have a StateTimeout of their own, and each get another: the page's time budget is what
	// ends a slow page (with Next), never a deadline spread over the whole of it.
	// ctx covers the bind and the listing alone; every bus request inside the loop takes each().
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	each := func() (context.Context, context.CancelFunc) {
		return context.WithTimeout(context.Background(), StateTimeout)
	}
	var keys []string
	var get func(key string) (*StateEntry, error)
	if s.PerNode {
		fam := c.family(s)
		if keys, err = fam.Keys(ctx); err != nil {
			return StatePage{}, err
		}
		buckets := map[string]jetstream.KeyValue{} // each node's bucket bound once
		get = func(key string) (*StateEntry, error) {
			node, err := pernode.NodeOf(key)
			if err != nil {
				return nil, err
			}
			kv, bound := buckets[node]
			if !bound {
				octx, ocancel := each() // the open is a bus request in the loop: its own deadline, never ctx's
				kv, err = fam.Open(octx, node)
				ocancel()
				if err != nil {
					return nil, err
				}
				buckets[node] = kv
			}
			if kv == nil {
				return nil, nil
			}
			gctx, gcancel := each()
			defer gcancel()
			e, err := kv.Get(gctx, key)
			if errors.Is(err, jetstream.ErrKeyNotFound) {
				return nil, nil
			}
			if err != nil {
				return nil, err
			}
			return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision(), At: e.Created().UTC()}, nil
		}
	} else {
		kv, err := c.bucket(ctx, s)
		if err != nil {
			return StatePage{}, err
		}
		if keys, err = keysOf(ctx, kv); err != nil {
			return StatePage{}, err
		}
		get = func(key string) (*StateEntry, error) {
			gctx, gcancel := each()
			defer gcancel()
			e, err := kv.Get(gctx, key)
			if errors.Is(err, jetstream.ErrKeyNotFound) {
				return nil, nil
			}
			if err != nil {
				return nil, err
			}
			return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision(), At: e.Created().UTC()}, nil
		}
	}
	sort.Strings(keys)
	page := StatePage{Entries: []StateEntry{}}
	started := PageClock()
	size := 0
	for _, k := range keys {
		if after != "" && k <= after {
			continue
		}
		if len(page.Entries) > 0 && PageClock().Sub(started) >= budget {
			page.Next = page.Entries[len(page.Entries)-1].Key
			break
		}
		e, err := get(k)
		if err != nil {
			return StatePage{}, err
		}
		if e == nil {
			continue
		}
		if size > 0 && size+len(k)+len(e.Value) > maxBytes {
			page.Next = page.Entries[len(page.Entries)-1].Key
			break
		}
		page.Entries = append(page.Entries, *e)
		size += len(k) + len(e.Value)
	}
	return page, nil
}

// StateKept is one value a bucket still keeps for a key: a put, or a delete (a purge reads as one).
type StateKept struct {
	Key      string          `json:"key"`
	Value    json.RawMessage `json:"value,omitempty"`
	Revision uint64          `json:"revision"`
	Op       string          `json:"op"`
	At       time.Time       `json:"at"`
}

// StateHistory is every value the bucket still keeps for one key, oldest first, or none (novox/hq ADR
// 0307). It is a read like StateGet: served to the owner and to every reader, and writes nothing.
func (c *Conn) StateHistory(module, name, key string) ([]StateKept, error) {
	s, err := c.issuedState(module, name)
	if err != nil {
		return nil, err
	}
	if err := checkKey(key); err != nil {
		return nil, err
	}
	if s.PerNode {
		return nil, perNodeRefusal(module, name, "history")
	}
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	kv, err := c.bucket(ctx, s)
	if err != nil {
		return nil, err
	}
	entries, err := kv.History(ctx, key)
	kept := []StateKept{}
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return kept, nil
	}
	if err != nil {
		return nil, err
	}
	for _, e := range entries {
		k := StateKept{Key: e.Key(), Revision: e.Revision(), Op: "delete", At: e.Created().UTC()}
		if e.Operation() == jetstream.KeyValuePut {
			k.Op = "put"
			k.Value = valueOf(e.Value())
		}
		kept = append(kept, k)
	}
	return kept, nil
}

// StateInfo is how full a state's bucket is against its caps (tools-are-how-a-module-is-spoken-to): a bundle's tools
// say fill against cap from it.
type StateInfo struct {
	State  string `json:"state"`
	Bucket string `json:"bucket"`
	// Bytes is what the bucket holds now, every kept value and marker counted; Keys how many keys it holds, a key
	// whose last value is a delete or expiry marker counted until the marker goes.
	Bytes uint64 `json:"bytes"`
	Keys  uint64 `json:"keys"`
	// MaxBytes and MaxValueBytes are the bucket's caps; -1 is none.
	MaxBytes      int64 `json:"max_bytes"`
	MaxValueBytes int32 `json:"max_value_bytes"`
	// History is how many values a key keeps; TTLSeconds the bucket's age for a value (0 none); KeyTTL whether a key may
	// carry its own time to live (issue 1952).
	History    int64 `json:"history"`
	TTLSeconds int64 `json:"ttl_seconds"`
	KeyTTL     bool  `json:"key_ttl"`
}

// StateInfo reads a state's bucket's fill and caps. It is a read like StateGet: served to the owner and every reader.
func (c *Conn) StateInfo(module, name string) (*StateInfo, error) {
	s, err := c.issuedState(module, name)
	if err != nil {
		return nil, err
	}
	if s.PerNode {
		return nil, perNodeRefusal(module, name, "fill")
	}
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	kv, err := c.bucket(ctx, s)
	if err != nil {
		return nil, err
	}
	status, err := kv.Status(ctx)
	if err != nil {
		return nil, err
	}
	b, ok := status.(*jetstream.KeyValueBucketStatus)
	if !ok || b.StreamInfo() == nil {
		return nil, fmt.Errorf("the bus answered no stream behind %s's state %s", module, name)
	}
	info := b.StreamInfo()
	return &StateInfo{State: name, Bucket: status.Bucket(), Bytes: info.State.Bytes, Keys: uint64(info.State.NumSubjects),
		MaxBytes: info.Config.MaxBytes, MaxValueBytes: info.Config.MaxMsgSize, History: status.History(),
		TTLSeconds: int64(status.TTL() / time.Second), KeyTTL: info.Config.AllowMsgTTL && status.LimitMarkerTTL() > 0}, nil
}

// StatePut writes one key, as the module, where the module keeps the state. It answers the revision.
func (c *Conn) StatePut(module, name, key string, value json.RawMessage) (uint64, error) {
	s, err := c.writable(module, name)
	if err != nil {
		return 0, err
	}
	if err := checkKey(key); err != nil {
		return 0, err
	}
	if len(value) == 0 || !json.Valid(value) {
		return 0, fmt.Errorf("a state value is JSON")
	}
	if field := credentialField(value); field != "" {
		return 0, fmt.Errorf("%s's %s.%s carries a field %q, which names a credential: no secret is kept in state, "+
			"sealed or not — a bucket is a stream, and a machine joining a year later reads it whole. Name the "+
			"secret and fetch it on request/reply (novox/hq ADR 0201, design 32 §10)", module, name, key, field)
	}
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	kv, err := c.bucket(ctx, s)
	if err != nil {
		return 0, err
	}
	return kv.Put(ctx, key, value)
}

// MinKeyTTL is the shortest time to live a key may be given: the bus counts a key's age in whole seconds (mesh-sdk
// process.MinTTL).
const MinKeyTTL = time.Second

// StateCreateTTL writes one key only when it has no value, with its own time to live (novox/hq ADR 0564 clauses 4 and
// 7, issue 1952; mesh-sdk process.State.CreateTTL): the bus removes it when ttl passes and every watch hears the
// removal as an expiry, so a bundle's wake needs no timer. A key that has a value is refused as ErrStateChanged. The
// bucket must take keys with their own time to live (a state declared `key-ttl`); one that does not is refused, naming
// it, before anything is written.
func (c *Conn) StateCreateTTL(module, name, key string, value json.RawMessage, ttl time.Duration) (uint64, error) {
	if ttl < MinKeyTTL {
		return 0, fmt.Errorf("%s's state %s, key %s: a time to live of %s is shorter than the bus keeps (%s)",
			module, name, key, ttl, MinKeyTTL)
	}
	return c.stateWrite(module, name, key, value, func(ctx context.Context, kv jetstream.KeyValue) (uint64, error) {
		if err := takesKeyTTL(ctx, kv, module, name); err != nil {
			return 0, err
		}
		return kv.Create(ctx, key, value, jetstream.KeyTTL(ttl))
	})
}

// StateRenewTTL writes one key whole with a fresh time to live, whatever it held (mesh-sdk process.State.RenewTTL): a
// time to live is set only when a key is created, so a live key is deleted first, and its watches hear the delete and
// then the new value.
func (c *Conn) StateRenewTTL(module, name, key string, value json.RawMessage, ttl time.Duration) (uint64, error) {
	rev, err := c.StateCreateTTL(module, name, key, value, ttl)
	if !errors.Is(err, ErrStateChanged) {
		return rev, err
	}
	if err := c.StateDelete(module, name, key); err != nil {
		return 0, err
	}
	return c.StateCreateTTL(module, name, key, value, ttl)
}

// takesKeyTTL refuses a bucket that does not allow a time to live per key, or leaves no marker when one ends: there the
// server would refuse the write, or the key would vanish and no watch would hear it.
func takesKeyTTL(ctx context.Context, kv jetstream.KeyValue, module, name string) error {
	status, err := kv.Status(ctx)
	if err != nil {
		return err
	}
	allows := status.LimitMarkerTTL() > 0
	if b, ok := status.(*jetstream.KeyValueBucketStatus); ok && b.StreamInfo() != nil {
		allows = allows && b.StreamInfo().Config.AllowMsgTTL
	}
	if !allows {
		return fmt.Errorf("%s's state %s does not take keys with their own time to live: its bucket %s allows none, or "+
			"leaves no marker a watch hears when one ends; %s declares the state `key-ttl` for it (novox/hq ADR 0564, "+
			"issue 1952)", module, name, status.Bucket(), module)
	}
	return nil
}

// StateDelete removes one key, as the module, where the module keeps the state. A key that was not
// there is not an error: what is asked for is that it is gone.
func (c *Conn) StateDelete(module, name, key string) error {
	s, err := c.writable(module, name)
	if err != nil {
		return err
	}
	if err := checkKey(key); err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	kv, err := c.bucket(ctx, s)
	if err != nil {
		return err
	}
	return kv.Delete(ctx, key)
}

// StateKeys is every key with a current value, sorted.
func (c *Conn) StateKeys(module, name string) ([]string, error) {
	s, err := c.issuedState(module, name)
	if err != nil {
		return nil, err
	}
	if s.PerNode {
		return c.perNodeKeys(s)
	}
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	defer cancel()
	kv, err := c.bucket(ctx, s)
	if err != nil {
		return nil, err
	}
	return keysOf(ctx, kv)
}

// keysOf is every key with a current value in one bucket, sorted.
func keysOf(ctx context.Context, kv jetstream.KeyValue) ([]string, error) {
	lister, err := kv.ListKeys(ctx)
	if err != nil {
		return nil, err
	}
	defer func() { _ = lister.Stop() }()
	keys := []string{}
	for k := range lister.Keys() {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys, nil
}

func (c *Conn) writable(module, name string) (StateIssued, error) {
	s, err := c.issuedState(module, name)
	if err != nil {
		return s, err
	}
	if !s.Writes {
		owner := name
		if dot := strings.LastIndex(name, "."); dot > 0 {
			owner = name[:dot]
		}
		return s, fmt.Errorf("%s reads %s and does not keep it: only %s's own instances write it (novox/hq ADR 0201)",
			module, name, owner)
	}
	return s, nil
}

// StateWatch hands deliver the current value of every key matching the pattern — none that is
// deleted — and then every change, in order (ADR 0201). It returns once the current values are
// delivered. **StateTimeout bounds the time without progress, not the whole set** (novox/hq issue
// 1726): each current value is a round trip to the bundle, so a state of a few hundred keys outlasts
// any fixed bound for all of them; the watch then failed at every attempt, the bundle dropped its
// handler, and the newest keys, handed over last, were never taken (the checker's owed, whose newest
// verdicts never reached mesh/check). It fails only when no current value arrives, or one is not
// taken, within StateTimeout; deliver is called from one goroutine, one change at a time, and an error from it is said
// and the watch goes on: state is not a queue, and the next change, or the next start, reads it again.
// The pattern is a key, with `*` for one name and `**` for the rest; empty is every key.
func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange) error) (stop func(), err error) {
	s, err := c.issuedState(module, name)
	if err != nil {
		return nil, err
	}
	filter := ">"
	if pattern != "" {
		parts := strings.Split(pattern, ".")
		for i, p := range parts {
			if p == "**" {
				if i != len(parts)-1 {
					return nil, fmt.Errorf("%q: `**` stands for the rest of a key, so it comes last", pattern)
				}
				parts[i] = ">"
				continue
			}
			if p != "*" && !stateKey.MatchString(p) {
				return nil, fmt.Errorf("%q is not a key pattern: names, `*` for one and `**` for the rest", pattern)
			}
		}
		filter = strings.Join(parts, ".")
	}
	if s.PerNode {
		return c.perNodeWatch(module, s, pattern, deliver)
	}
	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
	kv, err := c.bucket(ctx, s)
	cancel()
	if err != nil {
		return nil, err
	}
	return c.watchBucket(module, name, kv, filter, deliver)
}

// watchBucket is StateWatch over one bucket: the current values, then every change.
func (c *Conn) watchBucket(module, name string, kv jetstream.KeyValue, filter string, deliver func(StateChange) error) (stop func(), err error) {
	watching, stopWatching := context.WithCancel(context.Background())
	w, err := kv.Watch(watching, filter)
	if err != nil {
		stopWatching()
		return nil, err
	}
	var once sync.Once
	stop = func() {
		once.Do(func() {
			_ = w.Stop()
			stopWatching()
		})
	}
	current := make(chan struct{})
	// handed says a current value was handed over or passed by: progress, which restarts the bound.
	handed := make(chan struct{}, 1)
	progress := func() {
		select {
		case handed <- struct{}{}:
		default:
		}
	}
	go func() {
		initial := true
		for e := range w.Updates() {
			// A watch given up, or stopped, hands over nothing more, though values stay buffered in the
			// watcher: they would reach a watch id the bundle no longer holds (issue 774: 1257 orphan
			// changes after 51 failed watches).
			if watching.Err() != nil {
				break
			}
			if e == nil {
				// The end of what was there when the watch began.
				if initial {
					initial = false
					close(current)
				}
				continue
			}
			if initial {
				// Arrival from the bus is progress before the bundle is asked: the bound measures the current
				// values coming in, and restarts when the bundle has taken one.
				progress()
			}
			change := StateChange{State: name, Key: e.Key(), Revision: e.Revision(), Current: initial}
			switch e.Operation() {
			case jetstream.KeyValuePut:
				change.Op = "put"
				change.Value = valueOf(e.Value())
			default:
				// A deletion among the current values is a key that is not there: not handed over
				// (measured, research 024 — the server sends its marker among the initial values).
				if initial {
					progress()
					continue
				}
				change.Op = "delete"
				if e.Operation() == jetstream.KeyValuePurge {
					// The marker the server leaves when a key's time to live ends (Nats-Marker-Reason MaxAge,
					// read by the client as a purge): nothing else purges a module's state, since neither the
					// runtime nor mesh-sdk's State offers a purge (issue 1952).
					change.Op = "expire"
				}
			}
			if watching.Err() != nil {
				break
			}
			if err := deliver(change); err != nil {
				c.Logf("[mesh-tools] %s did not take %s %s of %s: %v; its next change, or its next start, reads it again",
					module, change.Op, change.Key, name, err)
			}
			if initial {
				progress()
			}
		}
		if initial {
			close(current)
		}
	}()
	bound := time.NewTimer(StateTimeout)
	defer bound.Stop()
	for {
		select {
		case <-current:
			return stop, nil
		case <-handed:
			if !bound.Stop() {
				select {
				case <-bound.C:
				default:
				}
			}
			bound.Reset(StateTimeout)
		case <-bound.C:
			stop()
			return nil, fmt.Errorf("the current values of %s did not arrive in %s: none was handed over in that time", name, StateTimeout)
		}
	}
}

func valueOf(raw []byte) json.RawMessage {
	if len(raw) == 0 || !json.Valid(raw) {
		b, _ := json.Marshal(string(raw))
		return b
	}
	return json.RawMessage(raw)
}

// credentialEndings are the ends of a field name that say its value is a credential. A guard against
// the ordinary mistake, not a determined one: a sealed value is plain text to anything inspecting it,
// so the rule is checked where it can be and said to be partial (ADR 0201).
var credentialEndings = []string{"password", "passwd", "secret", "token", "credential", "credentials",
	"authorization", "apikey", "privatekey", "accesskey", "cookie"}

// credentialField is the first field anywhere in a JSON value whose name says it is a credential, or "". The one shape a
// credential-shaped name may hold is a provider reference, an object of exactly `node` and `module`, both strings: a
// the assignment's binding of the provision "secret" to the vault (issue 1126) names who provides it, never its value. Any
// other shape under such a name is refused.
func credentialField(value json.RawMessage) string {
	var v any
	if json.Unmarshal(value, &v) != nil {
		return ""
	}
	var walk func(any) string
	walk = func(v any) string {
		switch t := v.(type) {
		case map[string]any:
			keys := make([]string, 0, len(t))
			for k := range t {
				keys = append(keys, k)
			}
			sort.Strings(keys)
			for _, k := range keys {
				norm := strings.NewReplacer("-", "", "_", "", " ", "").Replace(strings.ToLower(k))
				for _, end := range credentialEndings {
					if strings.HasSuffix(norm, end) && !providerReference(t[k]) {
						return k
					}
				}
				if found := walk(t[k]); found != "" {
					return found
				}
			}
		case []any:
			for _, x := range t {
				if found := walk(x); found != "" {
					return found
				}
			}
		}
		return ""
	}
	return walk(v)
}

// providerReference says whether v is exactly {"node": <name>, "module": <name>}, each a name as mesh-sdk names them.
func providerReference(v any) bool {
	m, ok := v.(map[string]any)
	if !ok || len(m) != 2 {
		return false
	}
	node, okN := m["node"].(string)
	module, okM := m["module"].(string)
	return okN && okM && referenceName.MatchString(node) && referenceName.MatchString(module)
}

// referenceName is a node's or a module's name as mesh-sdk names them: lower case, digits and dashes.
var referenceName = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*$`)