Merge pull request 'process: a module's own bus connection, subjects and durable consumer in one helper (issue 1242)' (#108) from fix/1242-bus-helper into main

go/process/account_test.go

@@ -257,3 +257,40 @@ func (h *heldSubscriber) deliver() {
 	h.mu.Unlock()
 	handle(&nats.Msg{Subject: subject})
 }
+
+// 1242: the module's own credential gives the names it reaches on the bus (its events' subjects, its own
+// stream's, its durable consumer's) and the options it dials with.
+func TestModuleAccountNamesAndDialOptions(t *testing.T) {
+	a := Account{URL: "nats://x:4222", Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}
+	for _, c := range []struct{ name, got, want string }{
+		{"event subject", a.EventSubject("put"), "mesh.mod.mesh-delivery-teller.event.put"},
+		{"own stream subject", a.StreamSubject("forge.x.pull"), "mesh.stream.mesh-delivery-teller.event.forge.x.pull"},
+		{"another module's stream subject", ModuleStreamSubject("mesh-delivery-checker", "check"), "mesh.stream.mesh-delivery-checker.event.check"},
+		{"durable consumer", a.EventConsumerName(), "n1_mesh-delivery-teller"},
+	} {
+		if c.got != c.want {
+			t.Errorf("%s: got %q, want %q", c.name, c.got, c.want)
+		}
+	}
+	if _, err := ReadBrokerAccount(func(string) string { return " " }); err == nil || !strings.Contains(err.Error(), BrokerFileEnv) {
+		t.Errorf("no broker file set: %v", err)
+	}
+	var o nats.Options
+	for _, opt := range a.DialOptions() {
+		if err := opt(&o); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if o.Name != "n1.mesh-delivery-teller" || o.User != "n1.mesh-delivery-teller" ||
+		o.MaxReconnect != -1 || !o.RetryOnFailedConnect || o.Secure {
+		t.Errorf("dial options: %+v", o)
+	}
+	a.Fingerprint = "sha256:ab"
+	o = nats.Options{}
+	for _, opt := range a.DialOptions() {
+		_ = opt(&o)
+	}
+	if !o.Secure || o.TLSConfig == nil {
+		t.Errorf("a fingerprint did not pin the connection")
+	}
+}

go/process/module.go

@@ -0,0 +1,95 @@
+package process
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// What a module's own principal reaches on the bus beyond a process account's queue (novox/hq issue 1242): its
+// own credential, its own stream and its durable consumer on the shared events stream. These helpers are the one
+// place a long-running part of a module (a daemon the module's delivery runs) learns them, so no part hand-writes
+// the connection, the subject spaces or the consumer's name.
+
+// EventsStream is the shared stream every module's events (mesh.mod.<module>.event.>) are on; a module's consumer
+// on it is the durable <node>_<module> the controller makes (mesh-controller internal/broker/nats.go
+// consumerStream, consumerDurable).
+const EventsStream = "EVENTS"
+
+// BrokerFileEnv names the environment variable holding the path of the module's own broker credential.
+const BrokerFileEnv = "MESH_BROKER_FILE"
+
+// ReadBrokerAccount reads the module's own credential from the file BrokerFileEnv names in getenv.
+func ReadBrokerAccount(getenv func(string) string) (Account, error) {
+	path := strings.TrimSpace(getenv(BrokerFileEnv))
+	if path == "" {
+		return Account{}, errors.New(BrokerFileEnv + " is not set, so there is no credential to hear the bus with")
+	}
+	return ReadAccount(path)
+}
+
+// DialOptions is the option list Dial connects with: named <node>.<module>, as the account's user on its own
+// inbox prefix, reconnecting without limit and retrying a first connect that fails, pinned to the account's
+// fingerprint when it carries one; opts follow, so a caller may add a handler or override one.
+func (a Account) DialOptions(opts ...nats.Option) []nats.Option {
+	all := []nats.Option{
+		nats.Name(a.Node + "." + a.Module),
+		nats.UserInfo(a.User, a.Password),
+		nats.CustomInboxPrefix(a.InboxPrefix()),
+		nats.MaxReconnects(-1),
+		nats.RetryOnFailedConnect(true),
+	}
+	if strings.TrimSpace(a.Fingerprint) != "" {
+		all = append(all, nats.Secure(PinnedToFingerprint(a.Fingerprint)))
+	}
+	return append(all, opts...)
+}
+
+// Dial connects as the module's own principal and opens JetStream on the connection. The caller closes the
+// connection.
+func (a Account) Dial(opts ...nats.Option) (*nats.Conn, jetstream.JetStream, error) {
+	nc, err := nats.Connect(a.URL, a.DialOptions(opts...)...)
+	if err != nil {
+		return nil, nil, fmt.Errorf("connecting to the bus: %w", err)
+	}
+	js, err := jetstream.New(nc)
+	if err != nil {
+		nc.Close()
+		return nil, nil, err
+	}
+	return nc, js, nil
+}
+
+// EventSubject is where the module's event lands on the shared events stream: mesh.mod.<module>.event.<event>.
+func (a Account) EventSubject(event string) string { return "mesh.mod." + a.Module + ".event." + event }
+
+// StreamSubject is where the module's event lands on a stream of its own: mesh.stream.<module>.event.<event>
+// (mesh-controller internal/broker/modstreams.go ModuleStreamSubject).
+func (a Account) StreamSubject(event string) string { return ModuleStreamSubject(a.Module, event) }
+
+// ModuleStreamSubject is where a module's event lands on that module's own stream, for a part that names another
+// module's stream (the indexer hears the forge bridge's, the checker's) without holding its credential.
+func ModuleStreamSubject(module, event string) string {
+	return "mesh.stream." + module + ".event." + event
+}
+
+// EventConsumerName is the module's durable consumer on a stream it is granted: <node>_<module>.
+func (a Account) EventConsumerName() string { return a.Node + "_" + a.Module }
+
+// EventConsumer binds the module's durable consumer on stream (EventsStream, or a stream of another module it is
+// granted).
+func (a Account) EventConsumer(ctx context.Context, js jetstream.JetStream, stream string) (jetstream.Consumer, error) {
+	return js.Consumer(ctx, stream, a.EventConsumerName())
+}
+
+// HearEvents pulls from cons, handing each message to handle, and says every error the bus reports while
+// hearing through logf ("hearing the bus: ..."). The caller stops the returned context.
+func HearEvents(cons jetstream.Consumer, handle func(jetstream.Msg), logf func(string, ...any)) (jetstream.ConsumeContext, error) {
+	return cons.Consume(handle, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) {
+		logf("hearing the bus: %v", err)
+	}))
+}