mesh-catalog/modules/nats/cmd/nats-tools/bus.go

package main

import (
	"context"
	"errors"
	"fmt"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"github.com/nats-io/nats.go/jetstream"
	"nats-tools/internal/busconn"
)

// LiveBus is the bus as the module's own credential reaches it. Nothing connects until a tool that changes the
// bus is called, so the tools that read (through the monitoring endpoint) answer as they did with no credential.
// NATS_TOOLS_CREDENTIAL names the credential (a word of the bundle's own: the runtime strips MESH_BROKER_FILE) (the process nats-tools's own secret `broker-nats-tools`); MESH_NATS_SERVER where the bus is as
// seen from this machine (default: its own port on loopback).
func LiveBus() Bus {
	return func(ctx context.Context) (jetstream.JetStream, func(), error) {
		file := env("NATS_TOOLS_CREDENTIAL", "")
		if file == "" {
			return nil, nil, errors.New("NATS_TOOLS_CREDENTIAL is not set, so this tool has no credential to reach the bus with (the module's manifest sets it)")
		}
		c, err := busconn.Read(file)
		if err != nil {
			return nil, nil, fmt.Errorf("the credential at %s (NATS_TOOLS_CREDENTIAL) cannot be read: %w", file, err)
		}
		nc, err := busconn.Connect(env("MESH_NATS_SERVER", "tls://127.0.0.1:4222"), c, "nats-tools")
		if err != nil {
			return nil, nil, err
		}
		js, err := jetstream.New(nc)
		if err != nil {
			nc.Close()
			return nil, nil, fmt.Errorf("JetStream on the bus: %w", err)
		}
		return js, nc.Close, nil
	}
}

// SaidByTheRuntime is the audit line: an event of this module, published through the runtime that launched it.
func SaidByTheRuntime(event string, body any) error { return stdio.Emit(event, body) }