package process import ( "context" "fmt" "strings" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/seatdef" ) // 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..`, the // seat mesh-delivery's `owed`). Its name on the bus is seatdef.KeptStream's, `SEAT__`, which a generated // contract carries as Stream: one name in one place. These are the one rule, for the part that makes the stream's // consumers (the grants part, from each module's `consumes`) and for every part that hears it, of the module's durable // on it. // keptStreamName is whether stream is a name seatdef.KeptStream gives: the bus's, with its prefix and no dot. func keptStreamName(stream string) bool { return strings.HasPrefix(stream, seatdef.KeptStreamPrefix) && len(stream) > len(seatdef.KeptStreamPrefix) && !strings.Contains(stream, ".") } // SeatEventConsumerConfig is the module's durable on a seat event's own stream as its maker makes it: _, 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 (a contract's Stream, 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) { if !keptStreamName(stream) { return nil, fmt.Errorf("%q is not the bus's name of a seat event's stream: a contract's Stream is %s_, "+ "with no dot (seatdef.KeptStream)", stream, seatdef.KeptStreamPrefix) } if handle == nil { return nil, fmt.Errorf("hearing %s needs a handler", stream) } if logf == nil { logf = func(string, ...any) {} } 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) } 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) } logf("events: the consumer %s on %s is heard", a.EventConsumerName(), stream) return cc, nil }