diff --git a/go/catalogue/seat_from_definition.go b/go/catalogue/seat_from_definition.go new file mode 100644 index 0000000..f67eb57 --- /dev/null +++ b/go/catalogue/seat_from_definition.go @@ -0,0 +1,57 @@ +package catalogue + +import ( + "sort" + + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +// SeatFromDefinition is a seat in this package's types from its definition (novox/hq ADR 0608 §1, ADR 0635 §1): what +// the seat list once held, now read from the definition the indexer keeps on the bus. The definition is resolved with +// its parts (seatdef.Resolve) as the indexer keeps it. A verb's input and answer have their named types in place, and a +// contribution kind given as data gets its checks (withShapeCode), which are code and never a definition's. +func SeatFromDefinition(d seatdef.Definition) Seat { + s := Seat{Name: d.Seat, Scope: d.Scope, Replicated: d.Replicated, Delivers: d.Delivers, + Accepts: seatdef.MessageNames(d.Accepts), Emits: seatdef.MessageNames(d.Events), PulledByKind: d.PulledByKind, + Yields: d.Yields, HeldByTheEngine: d.HeldByTheEngine, Decision: d.Decision} + for _, v := range d.Verbs { + s.Serves = append(s.Serves, Verb{Name: v.Name, Description: v.Description, + Input: seatdef.ResolveRefs(v.Input, d.Defs), Output: seatdef.ResolveRefs(v.Answer, d.Defs), + Optional: v.Optional, Replaces: v.Replaces}) + } + for _, r := range d.Receives { + rv := Receivable{Kind: r.Kind, Comment: r.Comment, Dirs: r.Dirs, Offered: r.Offered} + if r.Shape != nil { + sh := &Shape{Order: r.Shape.Order, Examples: r.Shape.Examples, Text: r.Shape.Text} + for _, f := range r.Shape.Places { + sh.Places = append(sh.Places, PlaceField{Field: f.Field, Values: f.Values}) + } + withShapeCode(d.Seat, r.Kind, sh) + rv.Shape = sh + } + s.Receives = append(s.Receives, rv) + } + for _, p := range d.Paths { + s.Paths = append(s.Paths, SeatPath{Path: p.Path, StepsBackFor: p.StepsBackFor}) + } + return s +} + +// SeatsFromDefinitions are the seats of a set of definitions, in reading order: the mesh's own, then a node's, each by +// name. A seat its holder declares itself (HolderDeclares) is not in the list, as the compiled list left it out. +func SeatsFromDefinitions(defs []seatdef.Definition) []Seat { + out := make([]Seat, 0, len(defs)) + for _, d := range defs { + if d.HolderDeclares { + continue + } + out = append(out, SeatFromDefinition(d)) + } + sort.SliceStable(out, func(i, j int) bool { + if (out[i].Scope == seatdef.ScopeMesh) != (out[j].Scope == seatdef.ScopeMesh) { + return out[i].Scope == seatdef.ScopeMesh + } + return out[i].Name < out[j].Name + }) + return out +} diff --git a/go/seatdef/refs.go b/go/seatdef/refs.go new file mode 100644 index 0000000..e3fc540 --- /dev/null +++ b/go/seatdef/refs.go @@ -0,0 +1,40 @@ +package seatdef + +import "strings" + +// ResolveRefs is a schema with every `{"$ref": "#/defs/"}` replaced by the definition's named type, as the seat +// list's readers need it (a verb's input and answer are one schema each, not a schema and its types). A reference to a +// name the definition does not define stays as it is; nesting deeper than 32 levels stops there. +func ResolveRefs(schema map[string]any, defs map[string]map[string]any) map[string]any { + if schema == nil { + return nil + } + m, _ := resolveValue(schema, defs, 0).(map[string]any) + return m +} + +func resolveValue(v any, defs map[string]map[string]any, depth int) any { + if depth > 32 { + return v + } + switch x := v.(type) { + case map[string]any: + if r, ok := x["$ref"].(string); ok { + if d, ok := defs[strings.TrimPrefix(r, "#/defs/")]; ok { + return resolveValue(d, defs, depth+1) + } + } + out := make(map[string]any, len(x)) + for k, e := range x { + out[k] = resolveValue(e, defs, depth+1) + } + return out + case []any: + out := make([]any, len(x)) + for i, e := range x { + out[i] = resolveValue(e, defs, depth+1) + } + return out + } + return v +} diff --git a/go/seatdef/refs_test.go b/go/seatdef/refs_test.go new file mode 100644 index 0000000..59576a4 --- /dev/null +++ b/go/seatdef/refs_test.go @@ -0,0 +1,31 @@ +package seatdef + +import ( + "reflect" + "testing" +) + +// A verb's schema has its named types in place (the seat list's readers take one schema, not a schema and its types). +func TestResolveRefsPutsNamedTypesInPlace(t *testing.T) { + defs := map[string]map[string]any{ + "id": {"type": "string"}, + "row": {"type": "object", "properties": map[string]any{"id": map[string]any{"$ref": "#/defs/id"}}}, + } + cases := []struct { + name string + schema, want map[string]any + }{ + {"nothing", nil, nil}, + {"a type nested in a type", map[string]any{"type": "array", "items": map[string]any{"$ref": "#/defs/row"}}, + map[string]any{"type": "array", "items": map[string]any{"type": "object", + "properties": map[string]any{"id": map[string]any{"type": "string"}}}}}, + {"a name it does not define stays", map[string]any{"$ref": "#/defs/gone"}, map[string]any{"$ref": "#/defs/gone"}}, + {"a list of schemas", map[string]any{"oneOf": []any{map[string]any{"$ref": "#/defs/id"}}}, + map[string]any{"oneOf": []any{map[string]any{"type": "string"}}}}, + } + for _, c := range cases { + if got := ResolveRefs(c.schema, defs); !reflect.DeepEqual(got, c.want) { + t.Errorf("%s: %v, want %v", c.name, got, c.want) + } + } +} diff --git a/go/seatlist/seatlist.go b/go/seatlist/seatlist.go new file mode 100644 index 0000000..68b8c50 --- /dev/null +++ b/go/seatlist/seatlist.go @@ -0,0 +1,227 @@ +// Package seatlist is the runtime's reader of the seat list (novox/hq ADR 0635 §1, §3): the seat definitions the +// indexer keeps on the bus, asked once through the owner's tool and followed on its event `seat-definition-changed`. +// The manifest validator, the node-engine, the tool runner, claude-code, the grants part and every other caller of +// `catalogue` get their seats here and give them to `catalogue` as an input; mesh-sdk compiles no seat list and imports +// no contracts module. +// +// **What it asks and hears.** The owner's tool Tool, with no seat for every trunk definition (the answer's shape is +// Listing) and with one seat for that seat's (the answer is the owner's own, read as Answer); the owner's event +// ChangedEvent, whose body names the seat and its source (Changed). Only trunk definitions are held: a pull request's are +// the indexer's to judge. Nothing polls and nothing waits on a timer: a reader that could not ask says so in its +// Status and reads again on the next event or on Reload, which the host module offers as a tool (steering it). +// +// **What it names.** The tool's and the event's names, which ADR 0635 §1 gives the owner. It names no module and no +// seat: the owner's module is the caller's (Config.Owner), the one its manifest's `invokes` lists. +package seatlist + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +const ( + // Tool is the owner's tool that answers a seat definition. + Tool = "indexer_seat_definition" + // ChangedEvent is the owner's event that says a definition moved. + ChangedEvent = "seat-definition-changed" + // SourceTrunk is the only source this reader holds. + SourceTrunk = "trunk" +) + +// Bus is what the reader needs of its host: a call of a tool by its key (`.`) and a subscription to an +// event pattern. StdioBus is the one a bundle has; the node-engine and the tool runner give theirs. +type Bus interface { + Ask(key string, body any) (json.RawMessage, error) + Subscribe(pattern string, handle func(key string, body json.RawMessage) error) error +} + +// Entry is one definition as the owner answers it: the seat, its source, the digest of the file, and the definition +// resolved with the parts it includes. It is the owner's SeatDefinitionEntry, read here by the fields it needs. +type Entry struct { + Seat string `json:"seat"` + Source string `json:"source"` + Digest string `json:"digest,omitempty"` + Definition json.RawMessage `json:"definition,omitempty"` + Unparsed string `json:"unparsed,omitempty"` +} + +// Listing is the owner's answer to the tool with no seat: every trunk definition and the revision of its state. +type Listing struct { + Revision uint64 `json:"revision"` + Definitions []Entry `json:"definitions"` +} + +// Answer is the owner's answer to the tool with one seat. +type Answer struct { + Seat string `json:"seat"` + Found bool `json:"found"` + Entry *Entry `json:"entry,omitempty"` + Revision uint64 `json:"revision"` +} + +// Changed is the event's body. +type Changed struct { + Seat string `json:"seat"` + Source string `json:"source"` + Digest string `json:"digest"` +} + +// Config is what a host gives Start. +type Config struct { + // Owner is the module that keeps the definitions, as the host's manifest `invokes` names it. + Owner string + // Bus is the host's way to the owner; StdioBus() for a bundle. + Bus Bus + // Bootstrap are the definitions known before the owner is heard (ADR 0635 §3); Bootstrapped() is the foundation's. + Bootstrap []seatdef.Definition + // OnChange, when set, is called after the held set changed, with the seats in reading order. It runs on the + // goroutine that heard the change: it returns quickly. + OnChange func([]catalogue.Seat) +} + +// Reader holds the seat list. +type Reader struct { + cfg Config + + mu sync.Mutex + defs map[string]seatdef.Definition // by seat + boot map[string]bool // seats held from the bootstrap, not yet from the owner +} + +// Start returns a reader holding the bootstrap, asks the owner for every definition and follows its event. A first +// ask the owner did not answer is no failure: the reader holds the bootstrap and reads again on the owner's next +// event or on Reload. Start returns the subscription's error alone, which the host may ask again by calling Start again +// on a new reader. +func Start(cfg Config) (*Reader, error) { + if cfg.Bus == nil { + return nil, errors.New("seatlist: no bus to read the seat list through") + } + if cfg.Owner == "" || strings.ContainsAny(cfg.Owner, ". \t") { + return nil, fmt.Errorf("seatlist: the owner %q is not a module name", cfg.Owner) + } + r := &Reader{cfg: cfg, defs: map[string]seatdef.Definition{}, boot: map[string]bool{}} + for _, d := range cfg.Bootstrap { + r.defs[d.Seat] = d + r.boot[d.Seat] = true + } + if err := cfg.Bus.Subscribe("*."+ChangedEvent, r.heard); err != nil { + return r, fmt.Errorf("seatlist: following %s: %w", ChangedEvent, err) + } + _ = r.Reload() + return r, nil +} + +// Seats are the held seats in reading order, as catalogue takes them. +func (r *Reader) Seats() []catalogue.Seat { + r.mu.Lock() + defer r.mu.Unlock() + return r.seatsLocked() +} + +// Definitions are the held definitions, by seat name. +func (r *Reader) Definitions() map[string]seatdef.Definition { + r.mu.Lock() + defer r.mu.Unlock() + out := make(map[string]seatdef.Definition, len(r.defs)) + for k, v := range r.defs { + out[k] = v + } + return out +} + +func (r *Reader) seatsLocked() []catalogue.Seat { + ds := make([]seatdef.Definition, 0, len(r.defs)) + for _, d := range r.defs { + ds = append(ds, d) + } + return catalogue.SeatsFromDefinitions(ds) +} + +// Reload asks the owner for every trunk definition and holds what it answers, a seat the owner holds replacing the +// bootstrap's. An answer that cannot be read is returned and changes nothing. +func (r *Reader) Reload() error { + raw, err := r.cfg.Bus.Ask(r.cfg.Owner+"."+Tool, map[string]any{}) + if err != nil { + return fmt.Errorf("seatlist: asking %s.%s: %w", r.cfg.Owner, Tool, err) + } + var l Listing + if err := json.Unmarshal(raw, &l); err != nil { + return fmt.Errorf("seatlist: %s.%s answered what is not a listing: %w", r.cfg.Owner, Tool, err) + } + parsed := map[string]seatdef.Definition{} + for _, e := range l.Definitions { + if e.Source != SourceTrunk || e.Unparsed != "" || len(e.Definition) == 0 { + continue + } + d, err := seatdef.Parse(e.Definition) + if err != nil { + return fmt.Errorf("seatlist: the definition of %s does not parse: %w", e.Seat, err) + } + parsed[d.Seat] = d + } + r.mu.Lock() + for seat := range r.defs { + if !r.boot[seat] { + delete(r.defs, seat) + } + } + for seat, d := range parsed { + r.defs[seat] = d + delete(r.boot, seat) + } + r.mu.Unlock() + r.changed() + return nil +} + +// heard takes the owner's event: a trunk definition moved, so that seat is asked again. An error is returned so the +// bus offers the event again. +func (r *Reader) heard(_ string, body json.RawMessage) error { + var c Changed + if err := json.Unmarshal(body, &c); err != nil || c.Seat == "" { + return nil + } + if c.Source != SourceTrunk { + return nil + } + raw, err := r.cfg.Bus.Ask(r.cfg.Owner+"."+Tool, map[string]any{"seat": c.Seat}) + if err != nil { + return err + } + var a Answer + if err := json.Unmarshal(raw, &a); err != nil { + return err + } + r.mu.Lock() + switch { + case !a.Found || a.Entry == nil || a.Entry.Source != SourceTrunk: + if !r.boot[c.Seat] { + delete(r.defs, c.Seat) + } + case a.Entry.Unparsed != "" || len(a.Entry.Definition) == 0: + // the definition that stands stays: a trunk's that does not parse is never written over it + default: + d, err := seatdef.Parse(a.Entry.Definition) + if err != nil { + r.mu.Unlock() + return fmt.Errorf("seatlist: the definition of %s does not parse: %w", c.Seat, err) + } + r.defs[c.Seat] = d + delete(r.boot, c.Seat) + } + r.mu.Unlock() + r.changed() + return nil +} + +func (r *Reader) changed() { + if r.cfg.OnChange != nil { + r.cfg.OnChange(r.Seats()) + } +} diff --git a/go/seatlist/seatlist_test.go b/go/seatlist/seatlist_test.go new file mode 100644 index 0000000..18750f8 --- /dev/null +++ b/go/seatlist/seatlist_test.go @@ -0,0 +1,129 @@ +package seatlist + +import ( + "encoding/json" + "errors" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +// fakeOwner is the definitions' owner: the tool answers from its table and the event is delivered by the test. +type fakeOwner struct { + table map[string]string // seat -> definition JSON, trunk + down bool + asked []string + handler func(string, json.RawMessage) error + pattern string +} + +func (f *fakeOwner) Ask(key string, body any) (json.RawMessage, error) { + f.asked = append(f.asked, key) + if f.down { + return nil, errors.New("no answer") + } + args, _ := body.(map[string]any) + if seat, _ := args["seat"].(string); seat != "" { + raw, ok := f.table[seat] + if !ok { + return json.Marshal(Answer{Seat: seat}) + } + return json.Marshal(Answer{Seat: seat, Found: true, Entry: &Entry{Seat: seat, Source: SourceTrunk, Definition: json.RawMessage(raw)}}) + } + l := Listing{Revision: 7} + for seat, raw := range f.table { + l.Definitions = append(l.Definitions, Entry{Seat: seat, Source: SourceTrunk, Definition: json.RawMessage(raw)}) + } + return json.Marshal(l) +} + +func (f *fakeOwner) Subscribe(pattern string, h func(string, json.RawMessage) error) error { + f.pattern, f.handler = pattern, h + return nil +} + +func (f *fakeOwner) say(seat, source string) error { + b, _ := json.Marshal(Changed{Seat: seat, Source: source, Digest: "d"}) + return f.handler("mesh-owner."+ChangedEvent, b) +} + +func def(seat, scope, verb string) string { + return `{"format":1,"seat":"` + seat + `","scope":"` + scope + `","grain":"seat-list","decision":"ADR 1",` + + `"verbs":[{"name":"` + verb + `","input":{"type":"object"}}]}` +} + +func names(r *Reader) string { + var out []string + for _, s := range r.Seats() { + out = append(out, s.Name) + } + return strings.Join(out, ",") +} + +// ADR 0635 §1, §3: the seat list comes from the owner on the bus, starting from the bootstrap, and follows its event. +func TestTheSeatListIsReadFromTheOwnerAndFollowsItsEvent(t *testing.T) { + boot, err := seatdef.Parse([]byte(def("boot-seat", "mesh", "go"))) + if err != nil { + t.Fatal(err) + } + owner := &fakeOwner{down: true, table: map[string]string{ + "node-b": def("node-b", "node", "put"), "mesh-a": def("mesh-a", "mesh", "get"), "boot-seat": def("boot-seat", "mesh", "other")}} + var told [][]catalogue.Seat + r, err := Start(Config{Owner: "mesh-owner", Bus: owner, Bootstrap: []seatdef.Definition{boot}, + OnChange: func(s []catalogue.Seat) { told = append(told, s) }}) + if err != nil { + t.Fatal(err) + } + if owner.pattern != "*."+ChangedEvent { + t.Fatalf("subscribed to %q", owner.pattern) + } + steps := []struct { + name string + act func() error + want string + err string + }{ + {"the owner has not answered: the bootstrap stands", func() error { return nil }, "boot-seat", ""}, + {"an event while the owner is down is offered again", func() error { return owner.say("mesh-a", SourceTrunk) }, "boot-seat", "no answer"}, + {"the owner answers: its seats in reading order, mesh seats first, the owner's definition replacing the bootstrap's", + func() error { owner.down = false; return r.Reload() }, "boot-seat,mesh-a,node-b", ""}, + {"a moved definition is asked again", func() error { + owner.table["mesh-a"] = def("mesh-a", "mesh", "renamed") + return owner.say("mesh-a", SourceTrunk) + }, "boot-seat,mesh-a,node-b", ""}, + {"a seat the owner holds no more goes", func() error { delete(owner.table, "node-b"); return owner.say("node-b", SourceTrunk) }, "boot-seat,mesh-a", ""}, + {"a pull request's definition is not held", func() error { owner.asked = nil; return owner.say("mesh-z", "pull") }, "boot-seat,mesh-a", ""}, + } + for _, s := range steps { + err := s.act() + if (err == nil) != (s.err == "") || (err != nil && !strings.Contains(err.Error(), s.err)) { + t.Fatalf("%s: error %v, want %q", s.name, err, s.err) + } + if got := names(r); got != s.want { + t.Fatalf("%s: seats %q, want %q", s.name, got, s.want) + } + } + if len(owner.asked) != 0 { + t.Fatalf("a pull request's event asked the owner: %v", owner.asked) + } + if v := r.Seats()[1].Serves[0]; v.Name != "renamed" || v.Input["type"] != "object" { + t.Fatalf("the moved definition's verb is %+v, want renamed ", v) + } + if len(told) != 3 { + t.Fatalf("OnChange was told %d times", len(told)) + } +} + +func TestStartRefusesWhatCannotRead(t *testing.T) { + for name, cfg := range map[string]Config{ + "no bus": {Owner: "m"}, + "an owner with a dot": {Owner: "m.tool", Bus: &fakeOwner{}}, + "no owner": {Bus: &fakeOwner{}}, + } { + if _, err := Start(cfg); err == nil { + t.Fatalf("%s: started", name) + } + } +} diff --git a/go/seatlist/stdiobus.go b/go/seatlist/stdiobus.go new file mode 100644 index 0000000..5ba2c4f --- /dev/null +++ b/go/seatlist/stdiobus.go @@ -0,0 +1,18 @@ +package seatlist + +import ( + "encoding/json" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +type stdioBus struct{} + +// StdioBus is the bus a bundle has: the runtime's ask and subscribe over its pipe. Call Start once Serve runs. +func StdioBus() Bus { return stdioBus{} } + +func (stdioBus) Ask(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) } + +func (stdioBus) Subscribe(pattern string, handle func(key string, body json.RawMessage) error) error { + return stdio.Subscribe(pattern, func(e stdio.Envelope) error { return handle(e.Key, e.Body) }) +}