Merge branch 'feat/47-G-asker7' into feat/47-G-join8

go/operatorask/operatorask.go

@@ -0,0 +1,111 @@
+// Package operatorask is a module's asking of a person through the seat that routes asks to the operator (design 47 G
+// A-101, §700, A-74): the seat's holder is found on the seat list, an ask is published under the module's own name, the
+// router's record of it is read for the warrant, and the router's `decided` is heard. It is shared code, so no module
+// keeps its own copy; it names no seat, verb or module: the generated contracts' names and types are its parameters.
+package operatorask
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// Caller calls a seat verb or module tool as `seat:<seat>.<verb>` or `<module>.<tool>` and returns its answer
+// (stdio.Ask is one; so is a process's calls).
+type Caller func(key string, body any) (json.RawMessage, error)
+
+// Contract is what the asker takes from the generated contracts: the verb that answers the seat list with its seats
+// (`Lists` is its call key, `ListBody` its input) and the asking seat and its state of the router's records.
+type Contract struct {
+	Lists    string
+	ListBody any
+	Seat     string
+	// StateAsks is the seat's state holding the router's record of each ask.
+	StateAsks string
+}
+
+// Bucket is the bucket of the router's records: the seat's state in the bucket of the seat's holder on the seat list.
+func Bucket(call Caller, c Contract) (string, error) {
+	raw, err := call(c.Lists, c.ListBody)
+	if err != nil {
+		return "", fmt.Errorf("who holds the seat %s is not asked of %s: %w", c.Seat, c.Lists, err)
+	}
+	var held struct {
+		Seats []struct {
+			Seat    string `json:"seat"`
+			Holders []struct {
+				Module string `json:"module"`
+			} `json:"holders"`
+		} `json:"seats"`
+	}
+	if err := json.Unmarshal(raw, &held); err != nil {
+		return "", fmt.Errorf("the answer of %s is not the seat list's: %w", c.Lists, err)
+	}
+	for _, s := range held.Seats {
+		if s.Seat == c.Seat && len(s.Holders) > 0 {
+			return s.Holders[0].Module + "_" + c.StateAsks, nil
+		}
+	}
+	return "", fmt.Errorf("no module holds the seat %s on the seat list, so the router's record of an ask is not read", c.Seat)
+}
+
+// Asker asks as one module. W is the generated contract's warrant type.
+type Asker[W any] struct {
+	Module   string
+	Contract Contract
+	// Call reaches the seat list's verb.
+	Call Caller
+	// Record reads one record of a bucket under the module's own name (process.Conn.Ask).
+	Record func(bucket, id string) (*stdio.StateEntry, error)
+	// Publish publishes a body de-duplicated by id: the same id twice is one message (stdio.SeatPublish).
+	Publish func(subject string, body any, id string) error
+	// Subscribe hears an event (stdio.Subscribe).
+	Subscribe func(pattern string, handle func(stdio.Envelope) error) error
+	// AskSubject and DecidedSubject are the generated contract's subjects for Module.
+	AskSubject, DecidedSubject func(module string) string
+}
+
+// Ask publishes the ask (the generated contract's Ask) under the module's own name, de-duplicated by its id.
+func (a *Asker[W]) Ask(id string, ask any) error {
+	if a.Publish == nil || a.AskSubject == nil {
+		return errors.New("the asker has no way to publish")
+	}
+	return a.Publish(a.AskSubject(a.Module), ask, id)
+}
+
+// Where is an ask's state as the router keeps it, and its warrant once there is one; a record not yet there is an empty
+// state. It is an error while the seat's holder or the record cannot be read.
+func (a *Asker[W]) Where(id string) (state string, warrant *W, err error) {
+	bucket, err := Bucket(a.Call, a.Contract)
+	if err != nil {
+		return "", nil, err
+	}
+	e, err := a.Record(bucket, id)
+	if err != nil || e == nil {
+		return "", nil, err
+	}
+	var rec struct {
+		State   string `json:"state"`
+		Warrant *W     `json:"warrant"`
+	}
+	if err := json.Unmarshal(e.Value, &rec); err != nil {
+		return "", nil, fmt.Errorf("the router's record of %s cannot be read: %w", id, err)
+	}
+	return rec.State, rec.Warrant, nil
+}
+
+// Hear hands handle each warrant the router says to this module; a handler's error leaves it for the bus to give again.
+func (a *Asker[W]) Hear(handle func(W) error) error {
+	if a.Subscribe == nil || a.DecidedSubject == nil {
+		return errors.New("the asker has no way to hear")
+	}
+	return a.Subscribe(a.DecidedSubject(a.Module), func(env stdio.Envelope) error {
+		var w W
+		if err := json.Unmarshal(env.Body, &w); err != nil {
+			return fmt.Errorf("%s holds what is no warrant: %w", env.Key, err)
+		}
+		return handle(w)
+	})
+}

go/operatorask/operatorask_test.go

@@ -0,0 +1,60 @@
+package operatorask
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+type warrant struct {
+	Ask    string `json:"ask"`
+	Choice string `json:"choice"`
+}
+
+// Design 47 G A-101: one asker finds the seat's holder, publishes, reads the warrant and hears decided.
+func TestTheAskerFindsTheHolderPublishesReadsTheWarrantAndHearsDecided(t *testing.T) {
+	var asked []string
+	call := func(key string, body any) (json.RawMessage, error) {
+		asked = append(asked, key)
+		return json.RawMessage(`{"seats":[{"seat":"other","holders":[{"module":"x"}]},{"seat":"router-seat","holders":[{"module":"holder"}]}]}`), nil
+	}
+	var bucket, id, subject string
+	var published any
+	var heard func(stdio.Envelope) error
+	a := &Asker[warrant]{
+		Module: "me", Contract: Contract{Lists: "seat:lister.list", ListBody: map[string]bool{"seats": true}, Seat: "router-seat", StateAsks: "asks"},
+		Call: call,
+		Record: func(b, i string) (*stdio.StateEntry, error) {
+			bucket, id = b, i
+			return &stdio.StateEntry{Value: json.RawMessage(`{"state":"chosen","warrant":{"ask":"a1","choice":"go"}}`)}, nil
+		},
+		Publish:        func(s string, body any, _ string) error { subject, published = s, body; return nil },
+		Subscribe:      func(_ string, h func(stdio.Envelope) error) error { heard = h; return nil },
+		AskSubject:     func(m string) string { return "ask." + m },
+		DecidedSubject: func(m string) string { return "decided." + m },
+	}
+	if err := a.Ask("a1", "q"); err != nil || subject != "ask.me" || published != "q" {
+		t.Errorf("Ask published %v on %q: %v", published, subject, err)
+	}
+	state, w, err := a.Where("a1")
+	if err != nil || state != "chosen" || w == nil || w.Choice != "go" || bucket != "holder_asks" || id != "a1" || len(asked) != 1 || asked[0] != "seat:lister.list" {
+		t.Errorf("Where = %q %+v %v, read %q %q after %v", state, w, err, bucket, id, asked)
+	}
+	var got []warrant
+	if err := a.Hear(func(w warrant) error { got = append(got, w); return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if err := heard(stdio.Envelope{Key: "decided.me", Body: []byte(`{"ask":"a1","choice":"go"}`)}); err != nil || len(got) != 1 || got[0].Ask != "a1" {
+		t.Errorf("heard %v: %v", got, err)
+	}
+	if err := heard(stdio.Envelope{Key: "decided.me", Body: []byte(`nope`)}); err == nil {
+		t.Errorf("a body that is no warrant was taken")
+	}
+	// No holder: nothing is read and the reason is said.
+	a.Call = func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"seats":[]}`), nil }
+	if _, _, err := a.Where("a1"); err == nil || !strings.Contains(err.Error(), "no module holds the seat router-seat") {
+		t.Errorf("with no holder Where answered %v", err)
+	}
+}

go/packages_list_test.go

@@ -26,7 +26,7 @@ var meshSDKPackages = []string{
 	".", "announce", "announcetest", "busarchive", "buscalls", "buscompose", "busstate", "bussnap", "busretry", "calladdress", "catalogue",
 	"declaration", "forgekey", "health", "internal/beside", "keys", "kvlimit", "membership", "motion",
 	"objectlink", "outward", "pernode", "process",
-	"sdkversion", "seal", "seatdef", "seatlist", "seatlist/seatlisttest", "stateobjects", "testbus", "views", "views/viewstest",
+	"sdkversion", "seal", "seatdef", "seatlist", "seatlist/seatlisttest", "operatorask", "stateobjects", "testbus", "views", "views/viewstest",
 }
 
 // clientSeats are the seats whose logic is their client's in mesh-contracts (novox/hq ADR 0627 §1, §4): mesh-sdk names

go/stdio.go

@@ -582,7 +582,8 @@ func Ask(key string, body any) (json.RawMessage, error) {
 }
 
 // TopicMatches says whether an event's type matches a pattern: `*` is one dot-separated segment, `#`
-// (or `**`) any number, as the TypeScript SDK matches.
+// (or `**`) any number, as the TypeScript SDK matches; the bus's own `>` as the last token matches one or more
+// further tokens (design 47 G A-101), so an adopted follower's `<seat>.<event>.>` hears every event on the seat's event.
 func TopicMatches(pattern, key string) bool {
 	return matchFrom(strings.Split(pattern, "."), 0, strings.Split(key, "."), 0)
 }
@@ -591,6 +592,9 @@ func matchFrom(p []string, pi int, k []string, ki int) bool {
 	if pi == len(p) {
 		return ki == len(k)
 	}
+	if p[pi] == ">" && pi == len(p)-1 {
+		return ki < len(k)
+	}
 	if p[pi] == "#" || p[pi] == "**" {
 		for skip := ki; skip <= len(k); skip++ {
 			if matchFrom(p, pi+1, k, skip) {

go/stdio_test.go

@@ -153,6 +153,18 @@ func TestABundleSubscribesAndAsksThroughTheRuntime(t *testing.T) {
 	if !TopicMatches("a.#", "a.b.c") || TopicMatches("a.*", "a.b.c") || !TopicMatches("#", "x") {
 		t.Error("topic matching differs from the TypeScript SDK's")
 	}
+	// Design 47 G A-101: the bus's `>` is one or more tokens at the end, and only there.
+	for pattern, want := range map[string]map[string]bool{
+		"s.rev.>": {"s.rev.a": true, "s.rev.a.b": true, "s.rev": false, "s.other.a": false},
+		"s.>.x":   {"s.a.x": false},
+		">":       {"a": true, "a.b": true},
+	} {
+		for key, ok := range want {
+			if TopicMatches(pattern, key) != ok {
+				t.Errorf("TopicMatches(%q, %q) = %v, want %v", pattern, key, !ok, ok)
+			}
+		}
+	}
 	// Issue 774 notes 84-87: a malformed key is refused before the runtime is asked.
 	for _, key := range []string{"seat:node-service-manager.stop@novox@g14", "seat:a.*", "postgres.list@", "postgres.list@a.b"} {
 		if _, err := Ask(key, nil); err == nil || strings.Contains(err.Error(), "Serve") {

go/views/views_test.go

@@ -719,6 +719,10 @@ func subViews(t *testing.T) {
 		if r.adopted["s.revisions"].local() != "s.revision-said.>" || len(heard) != 1 {
 			t.Errorf("the seat event %v reached the client %v", r.adopted, heard)
 		}
+		// A later revision, said with a further part (`<seat>.<event>.<part>`), reaches the follower by the runtime's own matching.
+		if pat := r.adopted["s.revisions"].local(); !stdio.TopicMatches(pat, "s.revision-said.catalog-revision-9") || !stdio.TopicMatches(pat, "s.revision-said.a.b") {
+			t.Errorf("the pattern %q does not hear a later catalog revision", pat)
+		}
 		f.WaitFor(context.Background(), "3")
 		ended <- true
 		if fd := fs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "reader.s.revisions" {