mesh-sdk/go/provider/provider.go

// Package provider is a providing module's half of a binding: it answers the provider's consumers itself, over the
// bus (novox/hq ADR 0349 clause 12, design 57 B1, issue 678).
//
// **Who binds the provider.** Each node's provisioner client records the needs of the modules handed to it from
// their plans, and the controller names in each plan, per required provision and local name, the provider that
// resolution chose: `provisioning.provisions` in `<node>.<module>` of the assignments mirror (issue 760). A recorded
// need names no provider (issue 677's note 7), so the plan's entry is where a bind is seen, as the vault sees it
// (mesh-vault pair.go). A consumer binds this provider when an entry names this machine and this module, and
// unbinds when the entry, or the assignment, goes. A consumer still filled the old way has no plan and is not seen
// here; the controller answers it as before (ADR 0349 clause 17).
//
// **One entry per consumer, in the provider's own bucket.** For each bind the helper composes the consumer's
// served values from the provider's offer (its config already composed into it, ADR 0202 §7) and the consumer's
// identity (`${consumer:as}`, `${consumer:as:dns}`, ADR 0202 §1), and writes one entry, `<node>.<module>.
// <provision>[.<local>]` of `<provider module>_bound`: the one statement both ends read. It never carries a
// secret. It names the credential the vault keeps for the pair, which the consumer's client asks the vault for.
//
// **The credential is the vault's** (ADR 0349 clause 8). The helper mints nothing and reads nothing of the
// provider's own secrets: it reads the value from the vault and hands it to the provider's adapter, which makes the
// login its software keeps. An offer whose software holds one credential (ADR 0158) is the vault's one value per
// provider and provision, `provision/<provider node>/<provision>`, with the provider's one user name; an offer with
// a login per consumer (ADR 0048, ADR 0158 §11, the default) is each consumer's own value,
// `pair/<provider node>/<provision>/<node>/<module>[/<local>]`, with the consumer's identity as its login (issue
// 677's notes 12 and 13). A credential the vault does not hold yet is waited for: nothing is made and no entry is
// written, so a consumer is never told of a login its provider has not made. A provider in the tool runner never asks
// the vault itself: its Vault is ConsumersFile, the file its machine's provisioner client fills (ADR 0373).
//
// **A multi-holder consumer gets no derived value** (ADR 0202 §10 as ADR 0349 clause 12 changes it): a consumer
// bound under a local name keeps several holders of the provision, each with its own login, while what the
// provider derives names one; its entry refuses it, naming both ends, and its client reports it.
//
// The helper speaks to the bus and the vault through the two interfaces below, so a provider wires it to its
// runtime and a test to a stand-in. Standard library only.
package provider

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

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

// AssignmentsBucket is the controller's assignments mirror, `<node>.<module>`, whose value carries the plan of a
// module handed to the provisioner.
const AssignmentsBucket = "mesh-controller_assignments"

// BucketOf is a providing module's own bucket of entries, one per consumer: `<module>_bound`, the module's state
// `bound`.
func BucketOf(module string) string { return module + "_bound" }

// Bus is what the helper needs of the bus: a key-value watch and the provider's writes to its own bucket.
type Bus interface {
	// Watch hands handle every current key and value of bucket, then each change as it comes, a removed key with a
	// nil value; it returns once the current values are handed, and the watch lasts until ctx ends.
	Watch(ctx context.Context, bucket string, handle func(key string, value []byte)) error
	Put(ctx context.Context, bucket, key string, value []byte) error
	Delete(ctx context.Context, bucket, key string) error
}

// Vault is how the provider reads a credential the vault keeps, by its name, opened.
type Vault interface {
	Value(ctx context.Context, name string) ([]byte, error)
}

// CredentialForm is which credential an offer's consumers present.
type CredentialForm int

const (
	// PerConsumer is a login of each consumer's own (ADR 0048, ADR 0158 §11): the default.
	PerConsumer CredentialForm = iota
	// Shared is the one credential the provider's software holds, shared with every consumer (ADR 0158).
	Shared
	// NoCredential is an offer whose consumers present none.
	NoCredential
)

// Offer is what the provider serves one provision with.
type Offer struct {
	// Serves are the values served to each consumer, the provider's config already composed into them; a string
	// may name the consumer's identity, `${consumer:as}` or `${consumer:as:dns}`.
	Serves map[string]any
	// Credential is the form of the credential its consumers present.
	Credential CredentialForm
	// User is the provider's one user name, for a Shared credential.
	User string
}

// Entry is one consumer's entry in the provider's bucket: the one statement both ends read. Never a secret.
type Entry struct {
	Provider       string `json:"provider"`
	ProviderModule string `json:"provider-module"`
	Provision      string `json:"provision"`
	Consumer       string `json:"consumer"`
	Module         string `json:"module"`
	Local          string `json:"local,omitempty"`
	// As is the consumer's identity, the mesh's (ADR 0202 §1).
	As string `json:"as,omitempty"`
	// User is the login the consumer presents: its identity, or the provider's one user name for a shared credential.
	User string `json:"user,omitempty"`
	// Values are the served values, composed for this consumer: what `${bound:<provision>:<key>}` reads.
	Values map[string]any `json:"values,omitempty"`
	// Credential is the name of the credential in the vault, which the consumer's client asks the vault for.
	Credential string `json:"credential,omitempty"`
	// Refused, when set, is why the consumer is not served, naming both ends.
	Refused string `json:"refused,omitempty"`
	// Contributes is what the consumer's module tells its provider about this provision (a database's name, its
	// extensions), from its plan's entry (issue 776). Handed to Make and Gone; never written to the bucket.
	Contributes map[string]any `json:"-"`
	// At and From are where the consumer's plan says the provider is and which machine it is (issue 788): the values
	// today's binding file gives that consumer. Written into Values as `at` and `from`; never alone to the bucket.
	At, From string `json:"-"`
	// Names are the names the controller composed for this consumer on this provision (issue 792): its own name, its
	// internal name, and one of each per local name it contributes under, keyed as its binding file names them.
	// Written into Values over anything the provider serves under the key; never alone to the bucket.
	Names map[string]string `json:"-"`
}

// Key is the entry's key in the provider's bucket: `<node>.<module>.<provision>`, and `.<local>` for a bind under a
// local name.
func (e Entry) Key() string {
	k := e.Consumer + "." + e.Module + "." + e.Provision
	if e.Local != "" {
		k += "." + e.Local
	}
	return k
}

// Consumer is what the provider's adapter makes: the consumer's entry and the credential the vault keeps for it.
type Consumer struct {
	Entry      Entry
	Credential []byte
}

// Options are the provider's.
type Options struct {
	// Node and Module are this provider: the machine it runs on and the module serving.
	Node, Module string
	// Offers are the provisions it serves, by name.
	Offers map[string]Offer
	Bus    Bus
	Vault  Vault
	// Make brings a consumer into being in the provider's software under the vault's credential. Asked on every
	// pass for every consumer answered, so it must leave one that is already as asked alone; a rotated credential
	// reaches the software this way.
	Make func(ctx context.Context, c Consumer) error
	// Gone, when set, hears that a consumer no longer binds the provider, before its entry is removed; an error
	// keeps the entry, and it is heard again on the next pass. Retiring what was kept for it is the provider's
	// (ADR 0230).
	Gone func(ctx context.Context, e Entry) error
	// Identity is the consumer's identity for an entry whose plan names none; ConsumerIdentity of the node and the
	// module when not set.
	Identity func(node, module string) string
	// Every is how often a pass runs without a change watched; a minute when not set.
	Every time.Duration
	// Logf says what a pass did that differs from the last; the standard log when not set.
	Logf func(format string, args ...any)
}

// Provider answers its consumers.
type Provider struct {
	opts Options

	mu          sync.Mutex
	assignments map[string][]byte
	entries     map[string][]byte
	said        map[string]string
	poke        chan struct{}
}

// New is a provider answering its consumers as opts say.
func New(opts Options) *Provider {
	if opts.Identity == nil {
		opts.Identity = ConsumerIdentity
	}
	if opts.Every <= 0 {
		opts.Every = time.Minute
	}
	if opts.Logf == nil {
		opts.Logf = func(f string, a ...any) { fmt.Printf("[provider] "+f+"\n", a...) }
	}
	return &Provider{opts: opts, assignments: map[string][]byte{}, entries: map[string][]byte{},
		said: map[string]string{}, poke: make(chan struct{}, 1)}
}

// What a pass did for one bind.
const (
	Answered = "answered"
	Refused  = "refused"
	Waiting  = "waiting"
	Failed   = "failed"
	Removed  = "removed"
)

// Outcome is what a pass did for one consumer's entry.
type Outcome struct {
	Key    string `json:"key"`
	Status string `json:"status"`
	Why    string `json:"why,omitempty"`
}

// Run watches the binds and the provider's own entries, and passes once at start, on every change and every
// Every, until ctx ends.
func (p *Provider) Run(ctx context.Context) error {
	if err := p.watch(ctx, BucketOf(p.opts.Module), p.entries); err != nil {
		return fmt.Errorf("watching %s: %w", BucketOf(p.opts.Module), err)
	}
	if err := p.watch(ctx, AssignmentsBucket, p.assignments); err != nil {
		return fmt.Errorf("watching %s: %w", AssignmentsBucket, err)
	}
	tick := time.NewTicker(p.opts.Every)
	defer tick.Stop()
	for {
		p.mu.Lock()
		assignments, entries := clone(p.assignments), clone(p.entries)
		p.mu.Unlock()
		if _, err := p.pass(ctx, assignments, entries); err != nil {
			p.say("pass", err.Error())
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-p.poke:
		case <-tick.C:
		}
	}
}

func (p *Provider) watch(ctx context.Context, bucket string, into map[string][]byte) error {
	return p.opts.Bus.Watch(ctx, bucket, func(key string, value []byte) {
		p.mu.Lock()
		if value == nil {
			delete(into, key)
		} else {
			into[key] = append([]byte(nil), value...)
		}
		p.mu.Unlock()
		if bucket == AssignmentsBucket {
			select {
			case p.poke <- struct{}{}:
			default:
			}
		}
	})
}

// Pass reads the binds and the provider's entries from the bus and answers once: each bind of this provider gets
// its entry, and each entry of this provider no bind asks for is removed. An assignment that cannot be read stops
// the pass before anything is removed. Run passes by itself; Pass is for a provider that passes when it chooses.
func (p *Provider) Pass(ctx context.Context) ([]Outcome, error) {
	watchCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	var mu sync.Mutex
	read := func(bucket string) (map[string][]byte, error) {
		got := map[string][]byte{}
		err := p.opts.Bus.Watch(watchCtx, bucket, func(key string, value []byte) {
			mu.Lock()
			defer mu.Unlock()
			if value == nil {
				delete(got, key)
			} else {
				got[key] = append([]byte(nil), value...)
			}
		})
		mu.Lock()
		defer mu.Unlock()
		return clone(got), err
	}
	entries, err := read(BucketOf(p.opts.Module))
	if err != nil {
		return nil, fmt.Errorf("reading %s: %w", BucketOf(p.opts.Module), err)
	}
	assignments, err := read(AssignmentsBucket)
	if err != nil {
		return nil, fmt.Errorf("reading %s: %w", AssignmentsBucket, err)
	}
	return p.pass(ctx, assignments, entries)
}

// planned is as much of an assignment as the helper reads: lenient, since the plan is the client's to act on.
type planned struct {
	Node         string `json:"node"`
	Module       string `json:"module"`
	Provisioning *struct {
		Provisions []struct {
			Provision      string            `json:"provision"`
			Local          string            `json:"local"`
			Provider       string            `json:"provider"`
			ProviderModule string            `json:"provider-module"`
			Contributes    map[string]any    `json:"contributes"`
			At             string            `json:"at"`
			From           string            `json:"from"`
			As             string            `json:"as"`
			Names          map[string]string `json:"names"`
		} `json:"provisions"`
	} `json:"provisioning"`
}

func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]byte) ([]Outcome, error) {
	var binds []Entry
	for _, key := range sortedKeys(assignments) {
		var a planned
		if err := json.Unmarshal(assignments[key], &a); err != nil || a.Node+"."+a.Module != key {
			return nil, fmt.Errorf("the assignment %s is not the controller's, so who binds %s on %s is not known; "+
				"nothing is answered or removed", key, p.opts.Module, p.opts.Node)
		}
		if a.Provisioning == nil {
			continue
		}
		for _, b := range a.Provisioning.Provisions {
			if b.Provider != p.opts.Node || b.ProviderModule != p.opts.Module {
				continue
			}
			binds = append(binds, Entry{Provider: p.opts.Node, ProviderModule: p.opts.Module,
				Provision: b.Provision, Consumer: a.Node, Module: a.Module, Local: b.Local,
				Contributes: b.Contributes, At: b.At, From: b.From, As: b.As, Names: b.Names})
		}
	}

	var out []Outcome
	want := map[string]bool{}
	for _, b := range binds {
		key := b.Key()
		want[key] = true
		o := p.answer(ctx, b, entries[key])
		out = append(out, o)
		p.say(key, o.Status+" "+o.Why)
	}
	for _, key := range sortedKeys(entries) {
		if want[key] {
			continue
		}
		var e Entry
		if err := json.Unmarshal(entries[key], &e); err != nil || e.Provider != p.opts.Node ||
			e.ProviderModule != p.opts.Module {
			continue // another machine's instance of this provider wrote it, or nothing of this helper's
		}
		if p.opts.Gone != nil {
			if err := p.opts.Gone(ctx, e); err != nil {
				o := Outcome{Key: key, Status: Failed, Why: "no longer bound, and the provider could not let it go: " +
					err.Error()}
				out = append(out, o)
				p.say(key, o.Status+" "+o.Why)
				continue
			}
		}
		if err := p.opts.Bus.Delete(ctx, BucketOf(p.opts.Module), key); err != nil {
			o := Outcome{Key: key, Status: Failed, Why: "no longer bound, and its entry could not be removed: " +
				err.Error()}
			out = append(out, o)
			p.say(key, o.Status+" "+o.Why)
			continue
		}
		out = append(out, Outcome{Key: key, Status: Removed})
		p.say(key, Removed)
		p.mu.Lock()
		delete(p.said, key)
		p.mu.Unlock()
	}
	return out, nil
}

// answer makes one bind's entry and writes it when it differs from what the bucket holds.
func (p *Provider) answer(ctx context.Context, e Entry, have []byte) Outcome {
	key := e.Key()
	offer, offered := p.opts.Offers[e.Provision]
	switch {
	case !offered:
		e.Refused = fmt.Sprintf("%s on %s is bound to %s on %s for %s, which %s does not serve", e.Module,
			e.Consumer, e.ProviderModule, e.Provider, e.Provision, e.ProviderModule)
	case e.Local != "" && derives(offer.Serves):
		e.Refused = fmt.Sprintf("%s on %s keeps several holders of %s (this one is %q), and %s on %s derives %s for "+
			"each consumer from its login: each holder has a login of its own and a consumer is told one value per "+
			"requirement, so the two ends would name different things (novox/hq ADR 0202 §10, ADR 0349 clause 12)",
			e.Module, e.Consumer, e.Provision, e.Local, e.ProviderModule, e.Provider,
			strings.Join(derivedKeys(offer.Serves), ", "))
	}
	if e.Refused != "" {
		return p.write(ctx, e, have, Outcome{Key: key, Status: Refused, Why: e.Refused})
	}

	// The plan's identity is the controller's, slug first; Identity only answers for an entry without one.
	if e.As == "" {
		e.As = p.opts.Identity(e.Consumer, e.Module)
	}
	values, err := ServedTo(offer.Serves, e.As)
	if err != nil {
		e.Refused = fmt.Sprintf("%s on %s serving %s to %s on %s: %v", e.ProviderModule, e.Provider, e.Provision,
			e.Module, e.Consumer, err)
		e.As = ""
		return p.write(ctx, e, have, Outcome{Key: key, Status: Refused, Why: e.Refused})
	}
	if e.At != "" || e.From != "" {
		if values == nil {
			values = map[string]any{}
		}
		if e.At != "" {
			values["at"] = e.At
		}
		if e.From != "" {
			values["from"] = e.From
		}
	}
	if len(e.Names) > 0 {
		if values == nil {
			values = map[string]any{}
		}
		for key, name := range e.Names {
			values[key] = name
		}
	}
	e.Values = values
	var credential []byte
	switch offer.Credential {
	case Shared:
		e.User = offer.User
		e.Credential = SharedName(e.Provider, e.Provision)
	case PerConsumer:
		e.User = e.As
		e.Credential = PairName(e.Provider, e.Provision, e.Consumer, e.Module, e.Local)
	}
	if e.Credential != "" {
		credential, err = p.opts.Vault.Value(ctx, e.Credential)
		if err != nil {
			return Outcome{Key: key, Status: Waiting, Why: fmt.Sprintf("the vault's %s: %v", e.Credential, err)}
		}
	}
	if p.opts.Make != nil {
		if err := p.opts.Make(ctx, Consumer{Entry: e, Credential: credential}); err != nil {
			why := err.Error()
			if len(credential) > 0 {
				why = strings.ReplaceAll(why, string(credential), "<credential>")
			}
			return Outcome{Key: key, Status: Failed, Why: "the provider could not make it: " + why}
		}
	}
	return p.write(ctx, e, have, Outcome{Key: key, Status: Answered})
}

func (p *Provider) write(ctx context.Context, e Entry, have []byte, o Outcome) Outcome {
	body, err := json.Marshal(e)
	if err != nil {
		return Outcome{Key: o.Key, Status: Failed, Why: err.Error()}
	}
	if sameJSON(have, body) {
		return o
	}
	if err := p.opts.Bus.Put(ctx, BucketOf(p.opts.Module), o.Key, body); err != nil {
		return Outcome{Key: o.Key, Status: Failed, Why: "its entry could not be written: " + err.Error()}
	}
	return o
}

// say logs what a key's pass said when it differs from the last.
func (p *Provider) say(key, what string) {
	p.mu.Lock()
	same := p.said[key] == what
	p.said[key] = what
	p.mu.Unlock()
	if !same {
		p.opts.Logf("%s: %s", key, strings.TrimSpace(what))
	}
}

// SharedName is the vault's one value per provider and provision (ADR 0158, issue 677).
func SharedName(provider, provision string) string { return "provision/" + provider + "/" + provision }

// PairName is the vault's value of one consumer's own login (ADR 0048, ADR 0158 §11, issue 677's note 12).
func PairName(provider, provision, node, module, local string) string {
	n := "pair/" + provider + "/" + provision + "/" + node + "/" + module
	if local != "" {
		n += "/" + local
	}
	return n
}

// identityUnusable is every character a consumer's identity does not hold (mesh-controller internal/catalogue
// identity.go, the one derivation).
var identityUnusable = regexp.MustCompile(`[^a-z0-9_]+`)

// ConsumerIdentity is what one module on one machine is called wherever it authenticates: `mesh_<node>_<module>`,
// lower case, anything but letters, digits and `_` written `_` (mesh-controller ConsumerIdentity). module is the
// identity source: the module's slug when it declares one, otherwise its name (ADR 0049).
func ConsumerIdentity(node, module string) string {
	clean := func(s string) string {
		return strings.Trim(identityUnusable.ReplaceAllString(strings.ToLower(s), "_"), "_")
	}
	return "mesh_" + clean(node) + "_" + clean(module)
}

// consumerFact is `${consumer:<fact>}` or `${consumer:<fact>:<alphabet>}`.
var consumerFact = regexp.MustCompile(`\$\{consumer:([a-z][a-z0-9-]*)(?::([a-z][a-z0-9-]*))?\}`)

// ServedTo fills an offer's served values for one consumer whose identity is as (ADR 0202 §1): `${consumer:as}` as
// it is, `${consumer:as:dns}` with `-` for `_`. Only strings carry placeholders; any other fact or alphabet is
// refused, never left standing.
func ServedTo(serves map[string]any, as string) (map[string]any, error) {
	if serves == nil {
		return nil, nil
	}
	out := make(map[string]any, len(serves))
	for _, key := range sortedKeys(serves) {
		text, ok := serves[key].(string)
		if !ok || !strings.Contains(text, "${consumer:") {
			out[key] = serves[key]
			continue
		}
		var failed error
		out[key] = consumerFact.ReplaceAllStringFunc(text, func(match string) string {
			parts := consumerFact.FindStringSubmatch(match)
			switch {
			case parts[1] == "as" && parts[2] == "":
				return as
			case parts[1] == "as" && parts[2] == "dns":
				return strings.ReplaceAll(as, "_", "-")
			}
			if failed == nil {
				failed = fmt.Errorf("the value served as %q says %s, and the mesh states only ${consumer:as} and "+
					"${consumer:as:dns}", key, match)
			}
			return match
		})
		if failed != nil {
			return nil, failed
		}
	}
	return out, nil
}

func derives(serves map[string]any) bool { return len(derivedKeys(serves)) > 0 }

func derivedKeys(serves map[string]any) []string {
	var out []string
	for _, k := range sortedKeys(serves) {
		if text, ok := serves[k].(string); ok && strings.Contains(text, "${consumer:") {
			out = append(out, k)
		}
	}
	return out
}

func clone(m map[string][]byte) map[string][]byte {
	out := make(map[string][]byte, len(m))
	for k, v := range m {
		out[k] = v
	}
	return out
}

func sameJSON(a, b []byte) bool {
	if a == nil || b == nil {
		return false
	}
	var x, y any
	if json.Unmarshal(a, &x) != nil || json.Unmarshal(b, &y) != nil {
		return false
	}
	ja, _ := json.Marshal(x)
	jb, _ := json.Marshal(y)
	return string(ja) == string(jb)
}

func sortedKeys[V any](m map[string]V) []string {
	if len(m) == 0 {
		return nil
	}
	out := make([]string, 0, len(m))
	for k := range m {
		out = append(out, k)
	}
	sort.Strings(out)
	return out
}

// StateBus is the Bus of a provider that runs as a module bundle over the runtime's stdio: the provider's own bucket,
// BucketOf(module), is the module's state `bound`, and AssignmentsBucket is the state it reads,
// `mesh-controller.assignments` (both declared in the module's manifest). Any other bucket is refused.
//
// A stdio state watch lasts as long as the process and cannot be ended (stdio.KeptState.Watch), so Watch honours the
// Bus contract by silencing instead: once ctx has ended the handler is never called again, though the runtime keeps
// sending changes that are dropped. A Pass, which watches twice per call, is therefore correct, only the dropped
// changes of its two spent watches remain.
func StateBus(module string) Bus { return stateBus{module: module} }

type stateBus struct{ module string }

func (b stateBus) state(bucket string) (stdio.KeptState, error) {
	switch bucket {
	case BucketOf(b.module):
		return stdio.State("bound"), nil
	case AssignmentsBucket:
		return stdio.State("mesh-controller.assignments"), nil
	}
	return stdio.KeptState{}, fmt.Errorf("%s is not a bucket this provider keeps or reads", bucket)
}

func (b stateBus) Watch(ctx context.Context, bucket string, handle func(key string, value []byte)) error {
	s, err := b.state(bucket)
	if err != nil {
		return err
	}
	return s.Watch("", func(ch stdio.StateChange) error {
		if ctx.Err() != nil {
			return nil
		}
		if ch.Op == "delete" {
			handle(ch.Key, nil)
		} else {
			handle(ch.Key, ch.Value)
		}
		return nil
	})
}

func (b stateBus) Put(_ context.Context, bucket, key string, value []byte) error {
	s, err := b.state(bucket)
	if err != nil {
		return err
	}
	_, err = s.Put(key, json.RawMessage(value))
	return err
}

func (b stateBus) Delete(_ context.Context, bucket, key string) error {
	s, err := b.state(bucket)
	if err != nil {
		return err
	}
	return s.Delete(key)
}