R-3: W1.12: a module hears a seat's event kept on its own stream through its durable

A seat's event kept apart from its family (the node-engine seat's running, kept last per subject; mesh-delivery's owed,
a stream with an age and a cap) had no helper: EventHub binds only the shared events stream, and dev-G-engine and
dev-G-forge each met it. process.SeatStreamName is the bus's name of a contract's Stream<E> (SEAT_<seat>_<event>), one
rule for its maker and its hearers; Account.SeatEventConsumerConfig the module's durable as the maker makes it (last
per subject where the event is kept so); Account.HearSeatEvents binds and pulls it, making nothing.

Builds: design 47 G §260, §24, §39, §146, §153, §199

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

go/process/event_hub_integration_test.go

@@ -65,3 +65,58 @@ func TestTheHubBindsALateConsumerAndSaysItIsLive(t *testing.T) {
 	}
 	hub.Stop()
 }
+
+// W1.12: a module hears a seat's event kept last per subject on its own stream through its durable, made by the maker
+// after the events were said, and is handed the last of each subject; without the consumer the call says so.
+func TestAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) {
+	nc, err := nats.Connect(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	kept, subjects := "node-engine.running", []string{"mesh.seat.node-engine.event.running.>"}
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: process.SeatStreamName(kept), Subjects: subjects, MaxMsgsPerSubject: 1}); err != nil {
+		t.Fatal(err)
+	}
+	for _, m := range []struct{ subject, fp string }{
+		{"mesh.seat.node-engine.event.running.ace.mesh-tools", "fp1"},
+		{"mesh.seat.node-engine.event.running.ace.mesh-tools", "fp2"},
+		{"mesh.seat.node-engine.event.running.g14.mesh-tools", "fp3"},
+	} {
+		if _, err := js.Publish(ctx, m.subject, []byte(m.fp)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	cred := process.Account{Node: "ace", Module: "mesh-grants"}
+	got := make(chan string, 4)
+	hear := func(m jetstream.Msg) { got <- m.Subject() + "=" + string(m.Data()); _ = m.Ack() }
+	if _, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf); err == nil {
+		t.Fatal("no consumer was made, and hearing said nothing")
+	}
+	if _, err := js.CreateOrUpdateConsumer(ctx, process.SeatStreamName(kept), cred.SeatEventConsumerConfig(subjects, true)); err != nil {
+		t.Fatal(err)
+	}
+	cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cc.Stop()
+	heard := map[string]bool{}
+	for len(heard) < 2 {
+		select {
+		case h := <-got:
+			heard[h] = true
+		case <-ctx.Done():
+			t.Fatalf("heard %v", heard)
+		}
+	}
+	if !heard["mesh.seat.node-engine.event.running.ace.mesh-tools=fp2"] || !heard["mesh.seat.node-engine.event.running.g14.mesh-tools=fp3"] {
+		t.Errorf("heard %v, want the last of each subject", heard)
+	}
+}

go/process/event_hub_test.go

@@ -35,4 +35,18 @@ func TestTheHubMatchesSubjectsAndRefusesOverlappingPatterns(t *testing.T) {
 	if err := h.Handle("mesh.mod.q.event.>", nop); err != nil {
 		t.Fatal(err)
 	}
+	// W1.12: a seat's event kept in a stream of its own is heard on one bus name and one durable, the maker's and the
+	// hearer's alike; an event kept last per subject is delivered from the last of each subject.
+	a := Account{Node: "ace", Module: "mesh-grants"}
+	if got := SeatStreamName("node-engine.running"); got != "SEAT_node-engine_running" {
+		t.Errorf("the bus's name of node-engine.running is %q", got)
+	}
+	cfg := a.SeatEventConsumerConfig([]string{"mesh.seat.node-engine.event.running.>"}, true)
+	if cfg.Durable != "ace_mesh-grants" || cfg.DeliverPolicy != jetstream.DeliverLastPerSubjectPolicy ||
+		cfg.AckPolicy != jetstream.AckExplicitPolicy || len(cfg.FilterSubjects) != 1 {
+		t.Errorf("the durable on a stream kept last per subject: %+v", cfg)
+	}
+	if cfg := a.SeatEventConsumerConfig(nil, false); cfg.DeliverPolicy != jetstream.DeliverAllPolicy {
+		t.Errorf("the durable on a stream kept with an age starts from its first message, got %v", cfg.DeliverPolicy)
+	}
 }

go/process/seatstream.go

@@ -0,0 +1,60 @@
+package process
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// A seat's event kept in a stream of its own (seatdef's `keep`: the last message of each subject, or a stream with an age
+// and a cap), not on the shared events stream (novox/hq design 47: the node-engine seat's `running.<node>.<module>`, the
+// seat mesh-delivery's `owed`). A generated contract names that stream as `<seat>.<event>` (Stream<E>, seatdef.KeptStream);
+// these are the one rule, for the part that makes the stream and its consumers (the grants part, from each module's
+// `consumes`) and for every part that hears it, of the stream's name on the bus and the module's durable on it.
+
+// SeatStreamPrefix starts the bus's name of every stream a seat's event is kept in apart from its family.
+const SeatStreamPrefix = "SEAT_"
+
+// SeatStreamName is the bus's name of the stream a contract names kept (`<seat>.<event>`): the bus takes no dot in a
+// stream's name, so `SEAT_<seat>_<event>`, each dot an underscore (`node-engine.running` is `SEAT_node-engine_running`).
+func SeatStreamName(kept string) string {
+	return SeatStreamPrefix + strings.ReplaceAll(kept, ".", "_")
+}
+
+// SeatEventConsumerConfig is the module's durable on a seat event's own stream as its maker makes it: <node>_<module>, as
+// on the shared stream, filtered to the subjects the module consumes, acknowledged one by one; for an event kept last
+// per subject it starts from the last message of each subject, never from now, so a consumer made after a subject was
+// last said still hears it, and two parts' consumers made at different times hold the same (design 47).
+func (a Account) SeatEventConsumerConfig(subjects []string, lastPerSubject bool) jetstream.ConsumerConfig {
+	cfg := jetstream.ConsumerConfig{Durable: a.EventConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, FilterSubjects: subjects}
+	if lastPerSubject {
+		cfg.DeliverPolicy = jetstream.DeliverLastPerSubjectPolicy
+	}
+	return cfg
+}
+
+// HearSeatEvents binds the module's durable on the stream of kept (a contract's Stream<E>) and pulls it, handing each
+// message to handle, which acknowledges it; every error the bus reports while hearing is said through logf. Like
+// EventHub it makes neither the stream nor the consumer: one the bus does not hold yet is this call's error, naming
+// both, and a later call binds it. The caller stops the returned context.
+func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, kept string, handle func(jetstream.Msg), logf func(string, ...any)) (jetstream.ConsumeContext, error) {
+	if handle == nil {
+		return nil, fmt.Errorf("hearing %s needs a handler", kept)
+	}
+	if logf == nil {
+		logf = func(string, ...any) {}
+	}
+	stream := SeatStreamName(kept)
+	cons, err := a.EventConsumer(ctx, js, stream)
+	if err != nil {
+		return nil, fmt.Errorf("the consumer %s on %s (the seat's stream %s) is not there: %w", a.EventConsumerName(), stream, kept, err)
+	}
+	cc, err := HearEvents(cons, handle, logf)
+	if err != nil {
+		return nil, fmt.Errorf("the consumer %s on %s (the seat's stream %s) is not heard: %w", a.EventConsumerName(), stream, kept, err)
+	}
+	logf("events: the consumer %s on %s (the seat's stream %s) is heard", a.EventConsumerName(), stream, kept)
+	return cc, nil
+}