mesh-delivery/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go

// Command mesh-delivery-forge-bridge is delivery's forge bridge (novox/hq ADR 0437 §4, issue 1235): it hears the forge
// module's pull request events and writes each as a pull message on its stream FORGE. It runs beside today's
// mesh-delivery and merges nothing. It serves no request subject and holds no timer: every reaction is an event
// (ADR 0437 §8); a refused act is tried again only on the next event touching its delivery.
package main

import (
	"context"
	"crypto/sha256"
	"crypto/tls"
	"crypto/x509"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-sdk/go/delivery"

	"mesh-delivery-forge-bridge/internal/bridge"
)

// eventsStream is the shared stream every module's events are on; the module's consumer on it is the durable
// <node>_<module> the controller makes (mesh-control internal/broker/nats.go consumerStream, consumerDurable).
const eventsStream = "EVENTS"

// The bridge's own stream FORGE takes `mesh.stream.<module>.event.<event>` (mesh-control internal/broker
// modstreams.go ModuleStreamSubject).
func forgeSubject(event string) string {
	return "mesh.stream." + delivery.ForgeBridgeModule + ".event." + event
}

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	if err := run(ctx, os.Getenv, os.Stderr); err != nil {
		fmt.Fprintln(os.Stderr, "mesh-delivery-forge-bridge:", err)
		os.Exit(1)
	}
}

// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate
// fingerprint to pin, the principal and its password (as the watchman reads it; issue 1242 moves this into the SDK).
type credential struct {
	URL         string `json:"url"`
	Fingerprint string `json:"fingerprint"`
	Node        string `json:"node"`
	Module      string `json:"module"`
	User        string `json:"user"`
	Password    string `json:"password"`
}

func readCredential(path string) (credential, error) {
	raw, err := os.ReadFile(path)
	if err != nil {
		return credential{}, err
	}
	var c credential
	if err := json.Unmarshal(raw, &c); err != nil {
		return credential{}, fmt.Errorf("the broker credential is not one: %w", err)
	}
	if c.URL == "" || c.User == "" {
		return credential{}, errors.New("the broker credential names no bus address or user")
	}
	return c, nil
}

// pinned accepts exactly the certificate with this SHA-256 and no other.
func pinned(want string) *tls.Config {
	return &tls.Config{
		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
		MinVersion:         tls.VersionTLS12,
		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
			if len(rawCerts) == 0 {
				return errors.New("the bus presented no certificate")
			}
			sum := sha256.Sum256(rawCerts[0])
			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
				return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got)
			}
			return nil
		},
	}
}

// delivered is the part of a JetStream message the bridge uses.
type delivered interface {
	Subject() string
	Data() []byte
	Headers() nats.Header
	Ack() error
	Term() error
	NakWithDelay(delay time.Duration) error
}

// retryDelay is how long the server waits before it delivers again an event the forge could not be asked about (the
// server's redelivery is the only clock; issue 1236/AC3).
const retryDelay = 10 * time.Second

// publisher writes to FORGE.
type publisher interface {
	Publish(ctx context.Context, subject string, payload []byte, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)
}

// handle turns one event into its message on FORGE and acknowledges it. An event it cannot read is terminated, never
// redelivered, and the log names its subject and why. A message it could not write stays unacknowledged: the server
// delivers it again (the server's ack wait is the only clock).
func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, logger *log.Logger) {
	msg, err := bridge.Convert(m.Subject(), m.Data(), trees)
	switch {
	case errors.Is(err, bridge.ErrRetry):
		logger.Printf("%v; heard again in %s", err, retryDelay)
		_ = m.NakWithDelay(retryDelay)
		return
	case errors.Is(err, bridge.ErrNoModules):
		logger.Printf("%v; no merged message", err)
		_ = m.Ack()
		return
	case errors.Is(err, bridge.ErrNotHandled):
		logger.Printf("%v; nothing written", err)
		_ = m.Ack()
		return
	case err != nil:
		logger.Printf("%v; the event is dropped", err)
		_ = m.Term()
		return
	}
	var opts []jetstream.PublishOpt
	// The forge's event id makes a redelivery after a lost acknowledgement the same message to the stream.
	if id := m.Headers().Get("x-event-id"); id != "" {
		opts = append(opts, jetstream.WithMsgID(id))
	}
	if _, err := pub.Publish(ctx, forgeSubject(msg.Event), msg.Data, opts...); err != nil {
		logger.Printf("%s: %s could not be written on FORGE, left for redelivery: %v", m.Subject(), msg.Event, err)
		return
	}
	_ = m.Ack()
}

// run is the program: it returns when ctx ends or the bus cannot be used.
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
	logger := log.New(logOut, "mesh-delivery-forge-bridge: ", log.LstdFlags)
	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
	if path == "" {
		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with")
	}
	cred, err := readCredential(path)
	if err != nil {
		return err
	}
	opts := []nats.Option{
		nats.Name(cred.Node + "." + cred.Module),
		nats.UserInfo(cred.User, cred.Password),
		nats.CustomInboxPrefix("_INBOX." + cred.User),
		nats.MaxReconnects(-1),
		nats.RetryOnFailedConnect(true),
	}
	if strings.TrimSpace(cred.Fingerprint) != "" {
		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
	}
	conn, err := nats.Connect(cred.URL, opts...)
	if err != nil {
		return fmt.Errorf("connecting to the bus: %w", err)
	}
	defer conn.Close()
	js, err := jetstream.New(conn)
	if err != nil {
		return err
	}
	cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module)
	if err != nil {
		return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, eventsStream, err)
	}
	cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, js, treesVia(ctx, callerOn(conn, cred.User)), logger) },
		jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
	if err != nil {
		return err
	}
	defer cc.Stop()
	<-ctx.Done()
	return nil
}