G §641: HearSeatEvents reads a stream kept last per subject whole at its start through an ordered consumer, then follows the durable, and says waiting then read

Builds: design 47 G §641, §556, A-19
Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

go/process/event_hub_integration_test.go

@@ -100,10 +100,15 @@ func TestAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) {
 	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, kept, cred.SeatEventConsumerConfig(subjects, true)); err != nil {
+	// The durable has gone past every message (it starts from now), so only the start read, an ordered consumer on the stream
+	// kept last per subject, gives the last of each subject (design 47 G §641, A-19).
+	durable := cred.SeatEventConsumerConfig(subjects, true)
+	durable.DeliverPolicy = jetstream.DeliverNewPolicy
+	if _, err := js.CreateOrUpdateConsumer(ctx, kept, durable); err != nil {
 		t.Fatal(err)
 	}
-	cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf)
+	var reads []string
+	cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf, process.OnRead(func(s string) { reads = append(reads, s) }))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -120,4 +125,19 @@ func TestAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) {
 	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)
 	}
+	if len(reads) != 2 || reads[0] != process.ReadWaiting || reads[1] != process.ReadDone {
+		t.Errorf("the read said %v, want waiting then read", reads)
+	}
+	// then the durable is followed
+	if _, err := js.Publish(ctx, "mesh.seat.node-engine.event.running.ace.mesh-tools", []byte("fp4")); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case h := <-got:
+		if h != "mesh.seat.node-engine.event.running.ace.mesh-tools=fp4" {
+			t.Errorf("the durable heard %s", h)
+		}
+	case <-ctx.Done():
+		t.Fatal("the durable was not followed after the read")
+	}
 }

go/process/seatstream.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync"
 
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -35,11 +36,63 @@ func (a Account) SeatEventConsumerConfig(subjects []string, lastPerSubject bool)
 	return cfg
 }
 
+// What HearSeatEvents says of its start read through OnRead (design 47 G §641, A-19): `waiting` when it begins the read of
+// a stream kept last per subject, `read` when the read has ended and the durable is followed.
+const (
+	ReadWaiting = "waiting"
+	ReadDone    = "read"
+)
+
+// HearOption is an option of HearSeatEvents.
+type HearOption func(*hearOptions)
+
+type hearOptions struct{ onRead func(state string) }
+
+// OnRead is the callback HearSeatEvents calls with ReadWaiting and then ReadDone around its start read; the module wires it
+// to its first views (design 47 G A-19), so this package imports nothing of views.
+func OnRead(f func(state string)) HearOption { return func(o *hearOptions) { o.onRead = f } }
+
+// readWhole hands handle every message an ordered consumer, made on stream for the one read, delivers from the last of each
+// subject filtered to subjects, until none is left pending (design 47 G §556): it ends when the message it delivered has
+// none after it, or at once when the consumer, made, holds none.
+func readWhole(ctx context.Context, js jetstream.JetStream, stream string, subjects []string, handle func(jetstream.Msg)) error {
+	cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: subjects, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy})
+	if err != nil {
+		return err
+	}
+	last := make(chan struct{})
+	var once sync.Once
+	cc, err := cons.Consume(func(m jetstream.Msg) {
+		handle(m)
+		if md, err := m.Metadata(); err == nil && md.NumPending == 0 {
+			once.Do(func() { close(last) })
+		}
+	})
+	if err != nil {
+		return err
+	}
+	defer cc.Stop()
+	if info := cons.CachedInfo(); info != nil && info.NumPending == 0 && info.Delivered.Consumer == 0 {
+		return nil
+	}
+	select {
+	case <-last:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
 // HearSeatEvents binds the module's durable on the stream (a contract's Stream<E>, as it stands) 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, stream string, handle func(jetstream.Msg), logf func(string, ...any)) (jetstream.ConsumeContext, error) {
+//
+// A stream the bus's own info says is kept last per subject is first read whole (design 47 G §641, §556, A-19): through an
+// ordered consumer made for that one read, on the subjects of the durable, to the last of each, its messages going to the
+// same handle; only then is the durable followed, so until the read ends the part hears none of it. OnRead says `waiting`
+// before the read and `read` after it.
+func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, stream string, handle func(jetstream.Msg), logf func(string, ...any), opts ...HearOption) (jetstream.ConsumeContext, error) {
 	if !keptStreamName(stream) {
 		return nil, fmt.Errorf("%q is not the bus's name of a seat event's stream: a contract's Stream<E> is %s<seat>_<event>, "+
 			"with no dot (seatdef.KeptStream)", stream, seatdef.KeptStreamPrefix)
@@ -50,10 +103,35 @@ func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, str
 	if logf == nil {
 		logf = func(string, ...any) {}
 	}
+	var o hearOptions
+	for _, f := range opts {
+		f(&o)
+	}
 	cons, err := a.EventConsumer(ctx, js, stream)
 	if err != nil {
 		return nil, fmt.Errorf("the consumer %s on %s is not there: %w", a.EventConsumerName(), stream, err)
 	}
+	st, err := js.Stream(ctx, stream)
+	if err != nil {
+		return nil, fmt.Errorf("the stream %s is not read: %w", stream, err)
+	}
+	if st.CachedInfo().Config.MaxMsgsPerSubject == 1 {
+		if o.onRead != nil {
+			o.onRead(ReadWaiting)
+		}
+		cfg := cons.CachedInfo().Config
+		subjects := cfg.FilterSubjects
+		if cfg.FilterSubject != "" {
+			subjects = append(subjects, cfg.FilterSubject)
+		}
+		if err := readWhole(ctx, js, stream, subjects, handle); err != nil {
+			return nil, fmt.Errorf("the start read of %s is not made: %w", stream, err)
+		}
+		if o.onRead != nil {
+			o.onRead(ReadDone)
+		}
+		logf("events: %s is read whole at its start", stream)
+	}
 	cc, err := HearEvents(cons, handle, logf)
 	if err != nil {
 		return nil, fmt.Errorf("the consumer %s on %s is not heard: %w", a.EventConsumerName(), stream, err)