From 65d570f11439a5a6b1cc5ef5708ce13e8c19eda7 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 17 Oct 2026 18:15:43 +0200 Subject: [PATCH] Vendor route-proxy's SDK so its provider builds offline (issue 1019) --- check-nested-modules.sh | 4 +- .../git.novox.be/novox/mesh-sdk/go/README.md | 19 + .../git.novox.be/novox/mesh-sdk/go/caller.go | 59 ++ .../git.novox.be/novox/mesh-sdk/go/log.go | 123 +++ .../novox/mesh-sdk/go/merge-check.sh | 20 + .../novox/mesh-sdk/go/provider/consumers.go | 72 ++ .../novox/mesh-sdk/go/provider/provider.go | 746 ++++++++++++++++++ .../novox/mesh-sdk/go/provision.go | 90 +++ .../git.novox.be/novox/mesh-sdk/go/seat.go | 153 ++++ .../git.novox.be/novox/mesh-sdk/go/state.go | 178 +++++ .../git.novox.be/novox/mesh-sdk/go/stdio.go | 522 ++++++++++++ modules/route-proxy/vendor/modules.txt | 4 + 12 files changed, 1988 insertions(+), 2 deletions(-) create mode 100644 modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/README.md create mode 100644 modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/caller.go create mode 100644 modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/log.go create mode 100644 modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh create mode 100644 modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/consumers.go create mode 100644 modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/provider.go create mode 100644 modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provision.go create mode 100644 modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/seat.go create mode 100644 modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go create mode 100644 modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go create mode 100644 modules/route-proxy/vendor/modules.txt diff --git a/check-nested-modules.sh b/check-nested-modules.sh index ce37d2f9..ee4b8905 100644 --- a/check-nested-modules.sh +++ b/check-nested-modules.sh @@ -9,8 +9,8 @@ race="${race:-}" # nested_modules vets and tests every Go module of its own below the root, `modules//go.mod` (novox/hq ADR 0420 # §5, issue 1025): `./...` stops at a nested go.mod, so a module that lives whole in this repository would have its # unit tests run by nobody. None is a pass. Each runs in its own directory, outside the vendor directory this -# repository's own packages are built from; how its dependencies are found offline is issue 1019's (the move of the -# first two modules in), and it is settled there. A failing module fails the check; the others are still run. +# repository's own packages are built from; a nested module vendors what the forge holds (its own `vendor/`, so the +# check runs offline), as this repository's root module does (issue 1019). A failing module fails the check; the others are still run. nested_modules() { nested=0 for gomod in modules/*/go.mod; do diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/README.md b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/README.md new file mode 100644 index 00000000..6a591b20 --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/README.md @@ -0,0 +1,19 @@ +# mesh-sdk for Go + +`stdio.Serve("", tools)` speaks MCP over stdio to the node's tool runtime, as the module the runtime +names (novox/hq ADR 0188, ADR 0193). A `Tool` is a name, a description, an argument schema and a +function; calls run concurrently. `stdio.CallerOf(args)` inside a tool is who called it: the bus user the runtime took from the subject the call arrived on, named in the call's `_meta`, never read from the arguments — or `stdio.Unproven` when the runtime named none (novox/hq issue 365). A tool named `.` is the seat's implementation. +`stdio.Emit(type, body)` publishes an event as the module, `stdio.Subscribe(pattern, handler)` receives the module's events (taken when the handler returns nil, offered again when it returns an error — the one rule is in the repository README, "Taking an event") and `stdio.Ask(key, body)` calls a tool, and `stdio.State(name)` gets, puts, deletes, lists, watches and reads the kept history of the module's declared state, and `stdio.Log(name)` appends to, reads, and measures a log of operations the module declared — all through the runtime (novox/hq ADR 0193, 0198, 0201, 0297, 0307). `stdio.Config(key)` and `stdio.Bound(provision, key)` read, and `stdio.WatchConfig` and `stdio.WatchBound` watch, the module's own config values and the non-secret values of what it is bound to, for what its manifest declares: the runtime asks its machine's provisioner client as the module it launched, never a module the code names. `stdio.Secret(name)` is refused while the module's code runs in the tool runner; its secrets reach it through its files (novox/hq ADR 0349 §14, ADR 0370). `provider` is a providing module's half of a binding (novox/hq ADR 0349 §12, issue 678): it sees each consumer that binds the provider in the plans of the assignments mirror, writes one entry per consumer to `_bound` (`..[.]`: served values composed with the consumer's identity, the login, and the name of the vault's credential, never a secret), hands the provider's adapter the vault's value (`provision//` shared, `pair////[/]` per consumer), refuses a derived value to a consumer bound under a local name, naming both ends, and removes the entry of one that unbinds; the bus and the vault are interfaces the provider wires, the vault as `provider.ConsumersFile`, the file its machine's provisioner client fills from `${consumers:}` with each bound consumer's credential (novox/hq ADR 0373), since a provider in the tool runner never asks the vault itself. Standard library only. + +## Installing + +``` +GOPRIVATE=git.novox.be go get git.novox.be/novox/mesh-sdk/go@latest +``` + +Releases are tagged `go/vX.Y.Z` in this repository, the subdirectory form Go's module system expects, +at the same version as the TypeScript package. + +What it holds, under ADR 0039's test: the stdio loop and the tool type, and `cli`, the local protocol +`mesh-cli` speaks to the node-engine on its machine (novox/hq ADR 0272): the socket's path and one request and one +answer; and `seal`, the mesh's sealed box (novox/hq ADR 0206: X25519, HKDF-SHA256, AES-256-GCM, `{v, eph, iv, tag, ct}`), byte for byte the box the licence manager made before it lived here and the TypeScript `@novox/mesh-sdk/seal` makes. Nothing else — no bus, no client, nothing of any module's. `go build ./example` and pipe a `tools/list` request into it. diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/caller.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/caller.go new file mode 100644 index 00000000..f9a108ac --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/caller.go @@ -0,0 +1,59 @@ +package stdio + +import ( + "reflect" + "regexp" + "sync" + "unsafe" +) + +// Unproven is the caller of a call the runtime named no caller for: one that arrived on the subject that names +// none, or from a runtime that names none (novox/hq issue 365). Said as a word, never left empty, so a record +// shows it was not proven. +const Unproven = "unproven" + +// MetaCaller is where in a `tools/call`'s `_meta` the runtime names the caller: the bus user that published the +// call, which the runtime takes from the subject the call arrived on — that subject's last token, which the +// bus lets each user publish with its own name alone (novox/hq issue 365, by ADR 0259 §3's precedent). The +// runtime writes `_meta`; a caller writes only `arguments`, so no caller can name another. +const MetaCaller = "mesh/caller" + +// busUser is the shape of a bus user's name: `.`, `person.`, `controller` — names whose +// parts are safe in a subject. Anything else the runtime named is not a caller and reads as Unproven. +var busUser = regexp.MustCompile(`^[A-Za-z0-9_-]+(\.[A-Za-z0-9_-]+)*$`) + +// calls holds the caller of every call while its tool runs, by the arguments map that call was handed. +var calls sync.Map // unsafe.Pointer → string + +func argsKey(args map[string]any) unsafe.Pointer { + return reflect.ValueOf(args).UnsafePointer() +} + +// CallerOf is who called the tool that was handed args, as the runtime proved it: the bus user that +// published the call, or Unproven. Read from the runtime, never from the arguments' contents — a `caller` +// argument names nobody. Valid while the tool runs; arguments no call was handed have no caller. +func CallerOf(args map[string]any) string { + if args == nil { + return Unproven + } + if c, ok := calls.Load(argsKey(args)); ok { + return c.(string) + } + return Unproven +} + +// callerIn is the caller a `tools/call`'s `_meta` names, or Unproven. +func callerIn(meta map[string]any) string { + if s, ok := meta[MetaCaller].(string); ok && busUser.MatchString(s) { + return s + } + return Unproven +} + +// runAs runs a tool with the caller held for its arguments for as long as it runs. +func runAs(t Tool, caller string, args map[string]any) (any, error) { + key := argsKey(args) + calls.Store(key, caller) + defer calls.Delete(key) + return t.Run(args) +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/log.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/log.go new file mode 100644 index 00000000..7b1030d8 --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/log.go @@ -0,0 +1,123 @@ +package stdio + +import ( + "encoding/json" + "fmt" + "time" +) + +// Log — a module's record of operations on the bus, kept as long as the log is (novox/hq ADR 0297). Not the +// current value: that is state, and a log's state is its fold. Each entry is one operation on one key, its +// sequence in the log is its identity, and the order of the log is the order of the operations. A module +// declares each log it owns in its manifest (`logs`) and names it here as it named it there; a log is its +// owner's alone, and the runtime refuses, with the reason, one it did not issue. + +// LogEntry is one operation on one key, as the log holds it. +type LogEntry struct { + Sequence uint64 `json:"sequence"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + At time.Time `json:"at"` +} + +// LogPage is one page of a read: its entries, in order, the sequence to read on from (0 when nothing +// follows) and the log's last sequence when it was read. +type LogPage struct { + Entries []LogEntry `json:"entries"` + Next uint64 `json:"next"` + Last uint64 `json:"last"` +} + +// LogInfo is how much a log holds: its entries and bytes, its cap in bytes, and its first and last sequence. +type LogInfo struct { + Entries uint64 `json:"entries"` + Bytes uint64 `json:"bytes"` + MaxBytes uint64 `json:"max_bytes"` + First uint64 `json:"first"` + Last uint64 `json:"last"` +} + +// KeptLog is one log this module owns. +type KeptLog struct{ name string } + +// Log is a log this module owns, by its local name. +func Log(name string) KeptLog { return KeptLog{name: name} } + +const logTimeout = 30 * time.Second + +func (l KeptLog) ask(verb string, params map[string]any) (json.RawMessage, error) { + c, err := live() + if err != nil { + return nil, err + } + params["log"] = l.name + return c.ask("mesh/log."+verb, params, logTimeout) +} + +func (l KeptLog) append(params map[string]any) (uint64, error) { + raw, err := l.ask("append", params) + if err != nil { + return 0, err + } + var appended struct { + Sequence uint64 `json:"sequence"` + } + return appended.Sequence, json.Unmarshal(raw, &appended) +} + +// Append writes an entry for key, whatever came before it, and returns its sequence. A key is one or more +// names of letters, digits, `-` and `_`, joined by `.`; a value is JSON of at most 256 KiB. A full log +// refuses the entry: it never drops an older one to make room. +func (l KeptLog) Append(key string, value any) (uint64, error) { + return l.append(map[string]any{"key": key, "value": value}) +} + +// AppendAt writes an entry for key only if the key's last entry is still at expect — the sequence the caller +// read, 0 for a key with no entry yet — and returns its sequence. When another entry for the key came in +// between, nothing is written and the error is ErrChanged, as for a lost State update: read again and +// decide again. +func (l KeptLog) AppendAt(key string, value any, expect uint64) (uint64, error) { + return l.append(map[string]any{"key": key, "value": value, "expect": expect}) +} + +// Read is the entries from the sequence from on, oldest first, at most limit of them (0 is the runtime's +// page, 100; it gives at most 1000). keys narrows it: a key, with `*` for one name and `>` for the rest; +// empty is every key. Read on from the page's Next until it is 0. +func (l KeptLog) Read(from uint64, keys string, limit int) (*LogPage, error) { + if limit < 0 { + return nil, fmt.Errorf("a read of %s asks for at least no entries: limit %d", l.name, limit) + } + params := map[string]any{"from": from} + if keys != "" { + params["keys"] = keys + } + if limit > 0 { + params["limit"] = limit + } + raw, err := l.ask("read", params) + if err != nil { + return nil, err + } + var p LogPage + return &p, json.Unmarshal(raw, &p) +} + +// Last is a key's last entry, or nil when the key has none. +func (l KeptLog) Last(key string) (*LogEntry, error) { + raw, err := l.ask("last", map[string]any{"key": key}) + if err != nil || len(raw) == 0 || string(raw) == "null" { + return nil, err + } + var e LogEntry + return &e, json.Unmarshal(raw, &e) +} + +// Info is how much the log holds and how much it may. +func (l KeptLog) Info() (*LogInfo, error) { + raw, err := l.ask("info", map[string]any{}) + if err != nil { + return nil, err + } + var i LogInfo + return &i, json.Unmarshal(raw, &i) +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh new file mode 100644 index 00000000..19a7ec01 --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh @@ -0,0 +1,20 @@ +#!/bin/sh +# mesh-check-toolchain: go +# +# The Go part of the SDK's own check (novox/hq ADR 0237 as amended, ADR 0272): the Go binding, run by the build +# seat in the mesh's Go toolchain, its own container, because the SDK's merge-check.sh declares it +# (`# mesh-check-also: go go/merge-check.sh`) — and by hand, from anywhere: `sh go/merge-check.sh`. +# +# Formatted, vetted, and its whole suite run: the stdio loop and the local protocol mesh-cli speaks to the +# node-engine. Standard library only, so nothing is fetched. +set -eu +cd "$(dirname "$0")" + +unformatted=$(gofmt -l .) +if [ -n "$unformatted" ]; then + echo "not gofmt'd:" + echo "$unformatted" + exit 1 +fi +go vet ./... +go test -count=1 ./... diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/consumers.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/consumers.go new file mode 100644 index 00000000..326d0a90 --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/consumers.go @@ -0,0 +1,72 @@ +package provider + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io/fs" + "os" +) + +// The provider's consumers' credentials, from the file its machine's provisioner client fills (novox/hq ADR 0373, +// issue 679). +// +// **The provider never asks the vault.** A provider in the tool runner has no account of its own, and the vault +// hands a secret only to a node's provisioner client or to a module under an account of its own (ADR 0349 clause +// 11, ADR 0370 clause 2). So the provider's template names `${consumers:}`, and its machine's client +// fills that file with one entry per consumer bound to it for that provision, each credential asked of the vault in +// the client's own name; the file is the provider's alone (mode 0600, owned). ConsumersFile is that file as the +// helper's Vault: it reads the file on each ask, so a consumer that binds or a credential the vault remakes reaches +// the provider at its next pass. + +// ConsumerCredential is one entry of the file: the consumer, the vault's name of its credential, and its value. +type ConsumerCredential struct { + // Consumer and Module are the consumer's machine and module, Local its local name for the provision, if any. + Consumer string `json:"consumer"` + Module string `json:"module"` + Local string `json:"local,omitempty"` + // Credential is the vault's name of the consumer's credential, PairName's. + Credential string `json:"credential"` + // Value is the credential itself. + Value string `json:"value"` +} + +// ConsumersFile answers a credential from the files the provider machine's client fills: one per provision whose +// template names `${consumers:}`, each a JSON list of ConsumerCredential. +type ConsumersFile struct { + Paths []string +} + +// Value is the credential named name, from the first file carrying it. A file not there yet is passed over; one not +// in the client's form is an error, never read as no consumers. +func (f ConsumersFile) Value(_ context.Context, name string) ([]byte, error) { + var missing []string + for _, at := range f.Paths { + raw, err := os.ReadFile(at) + if errors.Is(err, fs.ErrNotExist) { + missing = append(missing, at) + continue + } + if err != nil { + return nil, fmt.Errorf("the consumers' file %s: %w", at, err) + } + var entries []ConsumerCredential + if err := json.Unmarshal(raw, &entries); err != nil { + return nil, fmt.Errorf("the consumers' file %s is not the provisioner client's list of consumers", at) + } + for _, e := range entries { + if e.Credential == name { + if e.Value == "" { + return nil, fmt.Errorf("the consumers' file %s carries %s with no value", at, name) + } + return []byte(e.Value), nil + } + } + } + if len(missing) == len(f.Paths) { + return nil, fmt.Errorf("%s: no consumers' file is filled yet (%v)", name, f.Paths) + } + return nil, fmt.Errorf("%s is in no consumers' file: the vault holds none for it yet, or its machine's client "+ + "has not filled it", name) +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/provider.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/provider.go new file mode 100644 index 00000000..33bf61f7 --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/provider.go @@ -0,0 +1,746 @@ +// 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, issue 678): the provider module, not the controller, makes each bound +// consumer's entry. +// +// **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 `.` 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, `.. +// [.]` of `_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//`, 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////[/]`, 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/base64" + "encoding/json" + "errors" + "fmt" + "regexp" + "sort" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// AssignmentsBucket is the controller's assignments mirror, `.`, 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: `_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 +} + +// ConditionalBus is a Bus that can also delete a key only while it still holds what the caller read. The helper uses +// it when the Bus offers it, so a pass that lets a consumer go never removes an entry another instance of the +// provider wrote after the pass read it; a Bus without it deletes unconditionally. DeleteIf returns an error that +// is stdio.ErrChanged when the key holds something else now. +type ConditionalBus interface { + Bus + DeleteIf(ctx context.Context, bucket, key string, held []byte) 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. A consumer's plan entry that names a + // credential form of its own wins over it (the plan's `credential-form`). + 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::}` reads. + Values map[string]any `json:"values,omitempty"` + // VaultName is the name under which the vault keeps this consumer's credential, which the consumer's client asks + // the vault for: a name, never the credential. Its key is not "credential" because the runtime's state guard + // refuses any key ending in that word, whatever the value (issue 945). + VaultName string `json:"vault-name,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:"-"` + // CredentialForm and Name are from the consumer's plan entry (issue 825): the form of the credential the + // controller chose for this bind (`shared`, `pair`, `issued` or `given`) and the holder's name. When set, they + // win over the offer's form. Never written to the bucket. + CredentialForm, Name string `json:"-"` +} + +// Key is the entry's key in the provider's bucket: `..`, and `.` 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" + // Superseded is an entry no bind asks for any more that another instance of the provider wrote after this pass + // read it: it is left standing. + Superseded = "superseded" +) + +// 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, or that +// is not the one its key names, is skipped and said (an outcome of status Failed under the assignment's key, and a +// line of the log), and the entries of its consumer are left as they are, since a bind missed would read as an +// unbind; the pass goes on for the rest. Run passes by itself; Pass is for a provider that passes when it chooses. +// +// Over StateBus each call to Pass leaves two watches open for the life of the process, silenced once the call has +// ended (see StateBus): a provider that passes often should use Run, which watches once. +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"` + CredentialForm string `json:"credential-form"` + Name string `json:"name"` + } `json:"provisions"` + } `json:"provisioning"` +} + +func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]byte) ([]Outcome, error) { + var binds []Entry + var out []Outcome + unknown := map[string]bool{} // consumers whose assignment was skipped: their entries are not judged + for _, key := range sortedKeys(assignments) { + var a planned + if err := json.Unmarshal(assignments[key], &a); err != nil || a.Node+"."+a.Module != key { + o := Outcome{Key: key, Status: Failed, Why: fmt.Sprintf("the assignment %s is not the controller's, so "+ + "who it binds is not known; it is skipped, and its entries are left as they are", key)} + out = append(out, o) + p.say(key, o.Status+" "+o.Why) + unknown[key] = true + continue + } + 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, + CredentialForm: b.CredentialForm, Name: b.Name}) + } + } + + 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] || unknown[consumerOf(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.remove(ctx, key, entries[key]); err != nil { + if errors.Is(err, stdio.ErrChanged) { + o := Outcome{Key: key, Status: Superseded, Why: "another instance of the provider wrote its entry " + + "since this pass read it, so it stands"} + out = append(out, o) + p.say(key, o.Status+" "+o.Why) + continue + } + 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 +} + +// consumerOf is the `.` an entry's key starts with: its first two parts. +func consumerOf(key string) string { + parts := strings.SplitN(key, ".", 3) + if len(parts) < 3 { + return key + } + return parts[0] + "." + parts[1] +} + +// remove deletes an entry this instance read as held, and only while it is still as read when the Bus can say so. +func (p *Provider) remove(ctx context.Context, key string, held []byte) error { + if c, ok := p.opts.Bus.(ConditionalBus); ok { + return c.DeleteIf(ctx, BucketOf(p.opts.Module), key, held) + } + return p.opts.Bus.Delete(ctx, BucketOf(p.opts.Module), key) +} + +// 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 + // The plan's form wins; the offer's is the default. + // An offer that declares no credential has none whatever the plan says: the controller writes "pair" for it. + form := e.CredentialForm + if offer.Credential == NoCredential { + form = "" + } else if form == "" { + form = map[CredentialForm]string{Shared: "shared", PerConsumer: "pair"}[offer.Credential] + } + switch form { + case "shared": + e.User = offer.User + e.VaultName = SharedName(e.Provider, e.Provision) + case "pair": + e.User = e.As + e.VaultName = PairName(e.Provider, e.Provision, e.Consumer, e.Module, e.Local) + case "issued", "given": + // A per-holder value, as mesh-vault names it (perholder.go). + e.User = e.As + e.VaultName = HolderName(e.Provider, e.Provision, e.Consumer, e.Module) + } + if e.VaultName != "" { + credential, err = p.opts.Vault.Value(ctx, e.VaultName) + if err != nil { + return Outcome{Key: key, Status: Waiting, Why: fmt.Sprintf("the vault's %s: %v", e.VaultName, err)} + } + } + if p.opts.Make != nil { + if err := p.opts.Make(ctx, Consumer{Entry: e, Credential: credential}); err != nil { + why := err.Error() + why = redact(why, 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}) +} + +// redact takes a credential out of text in the forms it can come back in: as it is, in base64 (standard and URL, +// padded and not) and as a JSON string says it (escapes included). +func redact(text string, credential []byte) string { + if len(credential) == 0 { + return text + } + quoted, _ := json.Marshal(string(credential)) + forms := []string{string(credential), base64.StdEncoding.EncodeToString(credential), + base64.RawStdEncoding.EncodeToString(credential), base64.URLEncoding.EncodeToString(credential), + base64.RawURLEncoding.EncodeToString(credential), string(quoted[1 : len(quoted)-1])} + sort.SliceStable(forms, func(i, j int) bool { return len(forms[i]) > len(forms[j]) }) + for _, f := range forms { + text = strings.ReplaceAll(text, f, "") + } + return text +} + +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 } + +// HolderName is the vault's value of one holder's own credential, issued or given for it (mesh-vault perholder.go). +func HolderName(provider, provision, node, module string) string { + return "provision/" + provider + "/" + provision + "/" + node + "/" + module +} + +// 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__`, +// 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:}` or `${consumer::}`. +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 +} + +// DeleteIf removes the key only while it still holds what the caller read; it reads the key first, so another +// writer's change in that short while can still be lost. +func (b stateBus) DeleteIf(_ context.Context, bucket, key string, held []byte) error { + s, err := b.state(bucket) + if err != nil { + return err + } + cur, err := s.Get(key) + if err != nil { + return err + } + if cur == nil { + return nil + } + if !sameJSON(cur.Value, held) { + return fmt.Errorf("%s is not as it was read: %w", key, stdio.ErrChanged) + } + return s.Delete(key) +} + +func (b stateBus) Delete(_ context.Context, bucket, key string) error { + s, err := b.state(bucket) + if err != nil { + return err + } + return s.Delete(key) +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provision.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provision.go new file mode 100644 index 00000000..e4d65c7f --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provision.go @@ -0,0 +1,90 @@ +package stdio + +import ( + "encoding/json" + "errors" + "time" +) + +// A module's code reads its values at runtime (novox/hq ADR 0349 clause 14, ADR 0370, and the part of the feature 57 design that lets a running module read, rather than be handed in a file, its provisioned values): its config +// values and its bindings' values that are not secret, for what its manifest declares, and watches them. The +// runtime that launched this bundle asks its machine's provisioner client **as this bundle's module** — the module it +// launched, never a name this code passes — and the client answers only what that module declares, only while the +// mesh places it on that machine. A secret is refused while the module's code runs in the tool runner: the vault +// hands a secret only to a caller the bus proves, and the bus proves only the tool runner there, so a module's +// secrets reach it through its files until it runs under a bus account of its own (novox/hq issue 569). + +// ProvisionChange is what a watch hands over: the value now, or — once it is no longer answered, the module taken +// off the machine or the key no longer declared — Refused, why. +type ProvisionChange struct { + Kind string `json:"kind"` + Key string `json:"key"` + Value json.RawMessage `json:"value,omitempty"` + Refused string `json:"refused,omitempty"` +} + +const provisionTimeout = 30 * time.Second + +func provisionGet(kind, key string) (json.RawMessage, error) { + c, err := live() + if err != nil { + return nil, err + } + return c.ask("mesh/provision.get", map[string]any{"kind": kind, "key": key}, provisionTimeout) +} + +func provisionWatch(kind, key string, handle func(ProvisionChange) error) error { + if handle == nil { + return errors.New("a watch needs a handler") + } + c, err := live() + if err != nil { + return err + } + id := c.newID() + c.mu.Lock() + c.provisioned[id] = handle + c.mu.Unlock() + if _, err := c.askAs(id, "mesh/provision.watch", map[string]any{"kind": kind, "key": key}, provisionTimeout); err != nil { + c.mu.Lock() + delete(c.provisioned, id) + c.mu.Unlock() + return err + } + return nil +} + +// Config is one of this module's config values as mesh-config resolved it for this machine, as JSON: the key as +// its manifest declares it under `config`. A key it does not declare is refused, with the reason. +func Config(key string) (json.RawMessage, error) { return provisionGet("config", key) } + +// WatchConfig hands handle the config value now and then every change, in order, and returns once the value now is +// handled. The watch lasts as long as the process. +func WatchConfig(key string, handle func(ProvisionChange) error) error { + return provisionWatch("config", key, handle) +} + +// Bound is one value, not a secret, of what this module is bound to: the provision as its manifest requires it, +// and the key its provider answers (`${bound::}`). +func Bound(provision, key string) (json.RawMessage, error) { + return provisionGet("bound", provision+":"+key) +} + +// WatchBound is WatchConfig for a bound value. +func WatchBound(provision, key string, handle func(ProvisionChange) error) error { + return provisionWatch("bound", provision+":"+key, handle) +} + +// Secret is one of this module's secrets. Refused, with the reason, while the module's code runs in the tool +// runner (novox/hq ADR 0349 §11, ADR 0370 §2): its secrets reach it through its files. +func Secret(name string) ([]byte, error) { + raw, err := provisionGet("secret", name) + if err != nil { + return nil, err + } + var s string + if err := json.Unmarshal(raw, &s); err != nil { + return nil, errors.New("the runtime answered a secret that is not a string") + } + return []byte(s), nil +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/seat.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/seat.go new file mode 100644 index 00000000..aa2f144a --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/seat.go @@ -0,0 +1,153 @@ +package stdio + +import ( + "encoding/json" + "strings" + "time" +) + +// Seats that name their caller or their kind (novox/hq ADR 0259 §3): a module submits an accept or says an +// event under its own name or kind, takes a held seat's work, asks a proof and answers one, and reads a +// record a seat keeps for it. The runtime does each as the module, only as far as its membership lists, +// and refuses the rest with the reason. + +// Work is one piece of a seat's work this bundle takes: the subject it was submitted on, its body, and the +// headers the submitter's runtime set (x-source is the submitting module, x-event-id its id). +type Work struct { + Worker string `json:"worker,omitempty"` + Subject string `json:"subject"` + Body json.RawMessage `json:"body"` + Headers map[string]string `json:"headers,omitempty"` +} + +const seatTimeout = 30 * time.Second + +// SeatPublish submits an accept, or says an event, on a seat, de-duplicated by id: the same id twice is +// one message. An empty id is the runtime's to make. +func SeatPublish(subject string, body any, id string) error { + c, err := live() + if err != nil { + return err + } + _, err = c.ask("mesh/seat.publish", map[string]any{"subject": subject, "body": body, "id": id}, seatTimeout) + return err +} + +// SeatProve asks a proof — core request and reply, never kept — and returns its answer. +func SeatProve(subject string, body any) (json.RawMessage, error) { + c, err := live() + if err != nil { + return nil, err + } + return c.ask("mesh/seat.prove", map[string]any{"subject": subject, "body": body}, seatTimeout) +} + +// SeatAnswer answers every proof on a subject (a pattern, `*` one token) with what handle returns. +func SeatAnswer(subject string, handle func(subject string, body json.RawMessage) (any, error)) error { + c, err := live() + if err != nil { + return err + } + c.mu.Lock() + c.provers[subject] = handle + c.mu.Unlock() + _, err = c.ask("mesh/seat.answer", map[string]any{"subject": subject}, seatTimeout) + return err +} + +// SeatTake takes a held seat's work from a worker, handing each piece to handle: taken when it returns +// nil, offered again after an error. Make the work safe to do twice, or skip an x-event-id already done. +func SeatTake(worker string, handle func(Work) error) error { + c, err := live() + if err != nil { + return err + } + c.mu.Lock() + c.takers[worker] = handle + c.mu.Unlock() + _, err = c.ask("mesh/seat.take", map[string]any{"worker": worker}, seatTimeout) + return err +} + +// SeatRecord reads one key of a record a seat keeps for this module, under its own name, or nil. +func SeatRecord(bucket, key string) (*StateEntry, error) { + c, err := live() + if err != nil { + return nil, err + } + raw, err := c.ask("mesh/seat.record", map[string]any{"bucket": bucket, "key": key}, seatTimeout) + if err != nil || len(raw) == 0 || string(raw) == "null" { + return nil, err + } + var e StateEntry + return &e, json.Unmarshal(raw, &e) +} + +// KindHeld is one kind of a kinded bench and who holds it, with what it promises (novox/hq ADR 0259 §5). +type KindHeld struct { + Seat string `json:"seat"` + Kind string `json:"kind"` + Module string `json:"module"` + Node string `json:"node"` + Capabilities []string `json:"capabilities,omitempty"` +} + +// Promises says whether this kind's claim declares a capability. +func (k KindHeld) Promises(capability string) bool { + for _, c := range k.Capabilities { + if c == capability { + return true + } + } + return false +} + +// SeatKinds is every kind of the kinded benches this module uses or watches, as the controller issued +// them in its membership: the one account of which channel is which, never a channel's own word. +func SeatKinds() ([]KindHeld, error) { + c, err := live() + if err != nil { + return nil, err + } + raw, err := c.ask("mesh/seat.kinds", map[string]any{}, seatTimeout) + if err != nil { + return nil, err + } + var out []KindHeld + return out, json.Unmarshal(raw, &out) +} + +// Create writes a key only when it has none, and returns the revision; a key that has one is refused. +func (s KeptState) Create(key string, value any) (uint64, error) { + return s.revisionOf(s.ask("create", map[string]any{"key": key, "value": value})) +} + +// Update writes a key only when it is still at revision, and returns the new one; a key changed +// meanwhile is refused, so of two writers that read one value only the first writes. +func (s KeptState) Update(key string, value any, revision uint64) (uint64, error) { + return s.revisionOf(s.ask("update", map[string]any{"key": key, "value": value, "revision": revision})) +} + +func (s KeptState) revisionOf(raw json.RawMessage, err error) (uint64, error) { + if err != nil { + return 0, err + } + var put struct { + Revision uint64 `json:"revision"` + } + return put.Revision, json.Unmarshal(raw, &put) +} + +// subjectMatches is a subject against a pattern in the bus's wildcards: `*` one token, `>` the rest. +func subjectMatches(pattern, subject string) bool { + p, s := strings.Split(pattern, "."), strings.Split(subject, ".") + for i, part := range p { + if part == ">" { + return len(s) > i + } + if i >= len(s) || (part != "*" && part != s[i]) { + return false + } + } + return len(p) == len(s) +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go new file mode 100644 index 00000000..70e1b35c --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go @@ -0,0 +1,178 @@ +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 +// `.`) 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 (`.`). +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 +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go new file mode 100644 index 00000000..03db9bdc --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go @@ -0,0 +1,522 @@ +// Package stdio is the Novox Mesh SDK for Go (novox/hq ADR 0188, ADR 0193). +// +// A tools bundle is a process the node's runtime launches and speaks MCP over stdio to: +// `initialize`, `tools/list` once, `tools/call` per call, newline-framed JSON-RPC on stdin and +// stdout. This package is the loop, so a bundle writes its tools and nothing else. A tool named +// `.` is the module's implementation of that seat's verb; any other name is the +// module's own. +// +// No transport here: the bus, the subjects and the memberships are the runtime's (ADR 0039, 0175). +// What a bundle asks of the bus — emitting an event — it asks of the runtime over the same channel, +// and the runtime does it as the module (ADR 0193). stdout is the protocol; log to stderr. +package stdio + +import ( + "bufio" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "strconv" + "strings" + "sync" + "time" +) + +// Protocol is the MCP version this speaks; the runtime says the same. +const Protocol = "2025-03-26" + +// ServedAs is the word the runtime sets for a bundle it launches: the module it serves it as +// (ADR 0193). A bundle served under that name lists its own tools by their names. +const ServedAs = "MESH_SERVED_MODULE" + +// Tool is one tool a bundle serves. +type Tool struct { + Name string + Description string + // Input is the JSON schema of the arguments; nil is a tool that takes nothing. + Input map[string]any + // Run answers one call. Who called is CallerOf(args): the bus user the runtime proved, or Unproven. + Run func(args map[string]any) (any, error) +} + +type message struct { + ID *json.RawMessage `json:"id,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *struct { + Code int `json:"code"` + Message string `json:"message"` + } `json:"error,omitempty"` +} + +// channel is the one conversation with the runtime: what this bundle writes, serialised, and what +// it asked and waits for. +type channel struct { + mu sync.Mutex + enc *json.Encoder + next int + waiting map[string]chan message + listeners []listener + // subscribed numbers the listeners, so one whose subscribe failed is the one removed. + subscribed int + // watches are this bundle's state watches by the id each was asked with (ADR 0201). + watches map[string]func(StateChange) error + // provisioned are this bundle's watches of its values, by the id each was asked with (provision.go). + provisioned map[string]func(ProvisionChange) error + // takers and provers are this bundle's seat work and proof handlers, by worker and subject (ADR 0259). + takers map[string]func(Work) error + provers map[string]func(subject string, body json.RawMessage) (any, error) +} + +type listener struct { + n int + pattern string + handle func(Envelope) error +} + +// Envelope is one event as the mesh carries it: its type, the machine it came from, its body as JSON, +// and the ADR 0042 headers (x-event-id, x-source, x-node, x-time …). +type Envelope struct { + Key string `json:"key"` + Node string `json:"node"` + Body json.RawMessage `json:"body"` + Headers map[string]string `json:"headers,omitempty"` +} + +func (c *channel) say(m map[string]any) { + m["jsonrpc"] = "2.0" + c.mu.Lock() + defer c.mu.Unlock() + _ = c.enc.Encode(m) +} + +// ask sends the runtime a request and waits for its answer. +func (c *channel) ask(method string, params any, timeout time.Duration) (json.RawMessage, error) { + return c.askAs(c.newID(), method, params, timeout) +} + +func (c *channel) newID() string { + c.mu.Lock() + defer c.mu.Unlock() + c.next++ + return "bundle-" + strconv.Itoa(c.next) +} + +// askAs is ask under an id the caller chose, for a request whose id something is registered under +// before it is sent. +func (c *channel) askAs(id, method string, params any, timeout time.Duration) (json.RawMessage, error) { + c.mu.Lock() + answer := make(chan message, 1) + c.waiting[id] = answer + c.mu.Unlock() + c.say(map[string]any{"id": id, "method": method, "params": params}) + select { + case m := <-answer: + if m.Error != nil { + return nil, &refused{message: m.Error.Message, code: m.Error.Code} + } + return m.Result, nil + case <-time.After(timeout): + c.mu.Lock() + delete(c.waiting, id) + c.mu.Unlock() + return nil, fmt.Errorf("the runtime did not answer %s in %s", method, timeout) + } +} + +// refused is the runtime's error answer, as opposed to no answer: what it refused, it did not do. +type refused struct { + message string + code int +} + +func (r *refused) Error() string { return r.message } + +// Is makes a compare-and-set the runtime refused because another writer was first ErrChanged. +func (r *refused) Is(target error) bool { return target == ErrChanged && r.code == CodeChanged } + +// ErrChanged is a compare-and-set that lost: the key was made or changed by another writer first (novox/hq +// ADR 0259). Told apart by the runtime's error code, never by its words. +var ErrChanged = errors.New("the key was written by another writer first") + +// CodeChanged is the error code the runtime answers a lost compare-and-set with. +const CodeChanged = -32010 + +var ( + current *channel + currentMu sync.Mutex +) + +// Serve answers the runtime on stdin and stdout until stdin closes. An empty name serves as the +// module the runtime named in MESH_SERVED_MODULE. +func Serve(name string, tools []Tool) error { + return ServeOn(name, tools, os.Stdin, os.Stdout) +} + +// ServeOn is Serve over any pair of streams, for a test. +func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { + if name == "" { + name = os.Getenv(ServedAs) + } + if name == "" { + name = "bundle" + } + byName := map[string]Tool{} + for _, t := range tools { + byName[t.Name] = t + } + c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}, watches: map[string]func(StateChange) error{}, + provisioned: map[string]func(ProvisionChange) error{}, + takers: map[string]func(Work) error{}, provers: map[string]func(string, json.RawMessage) (any, error){}} + currentMu.Lock() + current = c + currentMu.Unlock() + scanner := bufio.NewScanner(in) + scanner.Buffer(make([]byte, 1<<20), 1<<24) + for scanner.Scan() { + var m message + if err := json.Unmarshal(scanner.Bytes(), &m); err != nil { + continue + } + // The runtime's answer to something this bundle asked. + if m.Method == "" && m.ID != nil { + var key string + if json.Unmarshal(*m.ID, &key) == nil { + c.mu.Lock() + answer, ok := c.waiting[key] + delete(c.waiting, key) + c.mu.Unlock() + if ok { + answer <- m + } + } + continue + } + id := any(nil) + if m.ID != nil { + id = *m.ID + } + switch m.Method { + case "initialize": + c.say(map[string]any{"id": id, "result": map[string]any{"protocolVersion": Protocol, + "capabilities": map[string]any{"tools": map[string]any{}}, + "serverInfo": map[string]any{"name": name, "version": "1"}}}) + case "notifications/initialized": + case "ping": + if id != nil { + c.say(map[string]any{"id": id, "result": map[string]any{}}) + } + case "tools/list": + listed := make([]map[string]any, 0, len(tools)) + for _, t := range tools { + schema := t.Input + if schema == nil { + schema = map[string]any{"type": "object", "properties": map[string]any{}} + } else if _, isSchema := schema["type"]; !isSchema { + schema = map[string]any{"type": "object", "properties": schema} + } + listed = append(listed, map[string]any{"name": t.Name, "description": t.Description, "inputSchema": schema}) + } + c.say(map[string]any{"id": id, "result": map[string]any{"tools": listed}}) + case "tools/call": + var p struct { + Name string `json:"name"` + Arguments map[string]any `json:"arguments"` + // Meta is the runtime's, never the caller's: who called, taken from the subject (CallerOf). + Meta map[string]any `json:"_meta"` + } + _ = json.Unmarshal(m.Params, &p) + t, ok := byName[p.Name] + if !ok { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": fmt.Sprintf("%s has no tool %s", name, p.Name)}}) + continue + } + // Beside the loop, never in it: a tool that emits waits for the runtime's answer, which + // arrives on the line this loop reads next — and calls do not queue behind a slow one. + go func(id any, t Tool, args map[string]any, caller string) { + if args == nil { + args = map[string]any{} + } + result, err := runAs(t, caller, args) + if err != nil { + c.say(map[string]any{"id": id, "result": map[string]any{"content": []map[string]any{{"type": "text", "text": err.Error()}}, "isError": true}}) + return + } + text, _ := json.Marshal(result) + c.say(map[string]any{"id": id, "result": map[string]any{"content": []map[string]any{{"type": "text", "text": string(text)}}}}) + }(id, t, p.Arguments, callerIn(p.Meta)) + case "mesh/event": + // One event the runtime read from this module's consumer (ADR 0198), answered once every + // handler it is for has taken it; an error is answered as one, so the runtime does not + // acknowledge it and the bus delivers it again. Beside the loop, like a tool call. + var p struct { + Envelope *Envelope `json:"envelope"` + } + if json.Unmarshal(m.Params, &p) != nil || p.Envelope == nil || p.Envelope.Key == "" { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/event carries no envelope"}}) + continue + } + c.mu.Lock() + matching := []listener{} + for _, l := range c.listeners { + if TopicMatches(l.pattern, p.Envelope.Key) { + matching = append(matching, l) + } + } + c.mu.Unlock() + go func(id any, env Envelope, ls []listener) { + for _, l := range ls { + if err := l.handle(env); err != nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + } + c.say(map[string]any{"id": id, "result": map[string]any{}}) + }(id, *p.Envelope, matching) + case "mesh/state": + // One change to a state this bundle watches (ADR 0201), named by the watch's id, answered + // once its handler has taken it. Beside the loop, like a tool call. + var p struct { + Watch string `json:"watch"` + Change *StateChange `json:"change"` + } + _ = json.Unmarshal(m.Params, &p) + c.mu.Lock() + handle := c.watches[p.Watch] + c.mu.Unlock() + if handle == nil || p.Change == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/state names no watch of this bundle"}}) + continue + } + go func(id any, change StateChange) { + if err := handle(change); err != nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + c.say(map[string]any{"id": id, "result": map[string]any{}}) + }(id, *p.Change) + case "mesh/provision": + // One change to a value of this module's that this bundle watches (provision.go), named by the + // watch's id, answered once its handler has taken it. Beside the loop, like a tool call. + var p struct { + Watch string `json:"watch"` + Change *ProvisionChange `json:"change"` + } + _ = json.Unmarshal(m.Params, &p) + c.mu.Lock() + handle := c.provisioned[p.Watch] + c.mu.Unlock() + if handle == nil || p.Change == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/provision names no watch of this bundle"}}) + continue + } + go func(id any, change ProvisionChange) { + if err := handle(change); err != nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + c.say(map[string]any{"id": id, "result": map[string]any{}}) + }(id, *p.Change) + case "mesh/work": + // One piece of a seat's work this bundle takes (novox/hq ADR 0259): taken by an answer that is + // not an error, offered again after one. Beside the loop, like a tool call. + var p struct { + Work *Work `json:"work"` + } + if json.Unmarshal(m.Params, &p) != nil || p.Work == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/work carries no work"}}) + continue + } + c.mu.Lock() + handle := c.takers[p.Work.Worker] + if handle == nil && len(c.takers) == 1 { + for _, h := range c.takers { + handle = h + } + } + c.mu.Unlock() + go func(id any, w Work) { + if handle == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": "nothing here takes " + w.Subject}}) + return + } + if err := handle(w); err != nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + c.say(map[string]any{"id": id, "result": map[string]any{}}) + }(id, *p.Work) + case "mesh/proof": + // One proof this bundle answers (novox/hq ADR 0259): what the handler returns is the answer; an + // error is answered in words, never as silence. + var p struct { + Subject string `json:"subject"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &p) + c.mu.Lock() + var handle func(string, json.RawMessage) (any, error) + for pattern, h := range c.provers { + if subjectMatches(pattern, p.Subject) { + handle = h + } + } + c.mu.Unlock() + go func(id any, subject string, body json.RawMessage) { + if handle == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": "nothing here answers " + subject}}) + return + } + out, err := handle(subject, body) + if err != nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + c.say(map[string]any{"id": id, "result": out}) + }(id, p.Subject, p.Body) + default: + if id != nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32601, "message": "no " + m.Method}}) + } + } + } + return scanner.Err() +} + +func live() (*channel, error) { + currentMu.Lock() + defer currentMu.Unlock() + if current == nil { + return nil, errors.New("before Serve: a bundle reaches the bus through the runtime that launched it") + } + return current, nil +} + +// Subscribe asks the runtime for this module's events whose type matches the pattern (`*` one +// segment, `#` any), and hands each to handle (ADR 0198). +// +// **Taking an event** (see the repository's README): handle returning nil takes the event — the +// runtime acknowledges it once every matching handler did; an error is the one way to have it offered +// again, after a few seconds and a bounded number of times, after which the bus gives up and the +// controller raises a condition. So return an error only when the work was not done, never for an +// answer that went right but read oddly; and since delivery is at-least-once, make the work safe to +// do twice, or skip an event whose x-event-id was already done. +// +// A subscribe the runtime refused leaves no handler behind, so asking again — as a module does until +// the runtime takes it — does not hand one event to the same handler twice. (One the runtime did not +// answer in time is kept, as the runtime may have bound it.) Call it from a goroutine +// started before Serve, or from a tool, once Serve runs. +func Subscribe(pattern string, handle func(Envelope) error) error { + c, err := live() + if err != nil { + return err + } + c.mu.Lock() + c.subscribed++ + n := c.subscribed + c.listeners = append(c.listeners, listener{n: n, pattern: pattern, handle: handle}) + c.mu.Unlock() + _, err = c.ask("mesh/subscribe", map[string]any{"pattern": pattern}, 30*time.Second) + // Removed only when the runtime answered that it refused: one that did not answer in time may + // have bound the consumer, and a handler removed then would have its events taken unhandled. + var no *refused + if errors.As(err, &no) { + c.mu.Lock() + for i, l := range c.listeners { + if l.n == n { + c.listeners = append(c.listeners[:i:i], c.listeners[i+1:]...) + break + } + } + c.mu.Unlock() + } + return err +} + +// Ask calls a tool on the mesh as this bundle's module, through the runtime (ADR 0198), and returns +// its result as JSON. The key is `.` as the runtime calls it. +func Ask(key string, body any) (json.RawMessage, error) { + c, err := live() + if err != nil { + return nil, err + } + return c.ask("mesh/ask", map[string]any{"key": key, "body": body}, 60*time.Second) +} + +// TopicMatches says whether an event's type matches a pattern: `*` is one dot-separated segment, `#` +// (or `**`) any number, as the TypeScript SDK matches. +func TopicMatches(pattern, key string) bool { + return matchFrom(strings.Split(pattern, "."), 0, strings.Split(key, "."), 0) +} + +func matchFrom(p []string, pi int, k []string, ki int) bool { + if pi == len(p) { + return ki == len(k) + } + if p[pi] == "#" || p[pi] == "**" { + for skip := ki; skip <= len(k); skip++ { + if matchFrom(p, pi+1, k, skip) { + return true + } + } + return false + } + if ki == len(k) || (p[pi] != "*" && p[pi] != k[ki]) { + return false + } + return matchFrom(p, pi+1, k, ki+1) +} + +// EmitOptions is the provenance an emitter may attach (ADR 0042). +type EmitOptions struct { + CausationID string + Schema string +} + +// Emit asks the runtime to publish an event as this bundle's module, and returns once the bus has +// accepted it (ADR 0193). The envelope and headers are the TypeScript SDK's exactly: the source and +// node come from MESH_MODULE and MESH_NODE, which the runtime sets. +func Emit(eventType string, body any, opts ...EmitOptions) error { + currentMu.Lock() + c := current + currentMu.Unlock() + if c == nil { + return errors.New("emit before Serve: a bundle emits through the runtime that launched it") + } + source, node := os.Getenv("MESH_MODULE"), os.Getenv("MESH_NODE") + if source == "" { + source = "unknown" + } + if node == "" { + node = "unknown" + } + headers := map[string]string{ + "x-event-id": eventID(), "x-source": source, "x-node": node, + "x-time": time.Now().UTC().Format(time.RFC3339Nano), "content-type": "application/json", + } + for _, o := range opts { + if o.CausationID != "" { + headers["x-causation-id"] = o.CausationID + } + if o.Schema != "" { + headers["x-schema"] = o.Schema + } + } + _, err := c.ask("mesh/publish", map[string]any{"key": eventType, "node": node, "body": body, "headers": headers}, 30*time.Second) + return err +} + +// eventID is a random UUID, version 4, as the TypeScript SDK's randomUUID. +func eventID() string { + var b [16]byte + _, _ = rand.Read(b[:]) + b[6] = (b[6] & 0x0f) | 0x40 + b[8] = (b[8] & 0x3f) | 0x80 + h := hex.EncodeToString(b[:]) + return h[0:8] + "-" + h[8:12] + "-" + h[12:16] + "-" + h[16:20] + "-" + h[20:32] +} diff --git a/modules/route-proxy/vendor/modules.txt b/modules/route-proxy/vendor/modules.txt new file mode 100644 index 00000000..abae1a04 --- /dev/null +++ b/modules/route-proxy/vendor/modules.txt @@ -0,0 +1,4 @@ +# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261016174426-a4e7f480b3b0 +## explicit; go 1.22 +git.novox.be/novox/mesh-sdk/go +git.novox.be/novox/mesh-sdk/go/provider