mesh-sdk/go/process/module.go

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) {
	OwnBuild()
	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) }

// BusSubject is where the module's event lands on a stream it asked for by template: mesh.bus.<module>.<local>.<event>,
// <local> being the stream's local name in the module's `requires` (the bus module's keeper, PlanOf; A-41).
func (a Account) BusSubject(local, event string) string {
	return "mesh.bus." + a.Module + "." + local + "." + 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)
	}))
}