Give a module's own credential its dial, subjects and durable consumer in mesh-sdk, so a part no longer hand-writes them (issue 1242)

go/process/module.go

@@ -0,0 +1,91 @@
+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 "mesh.stream." + a.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)
+	}))
+}