mesh-controller/internal/link/bus.go

package link

import (
	"context"
	"fmt"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
	"github.com/nats-io/nats.go"
)

// Bus is what the controller needs of the mesh's bus, **in the mesh's own words rather than a
// transport's** (novox/hq ADR 0116 step 3).
//
// Until now every one of these functions took an `*amqp.Channel`, so the transport reached every
// caller and swapping it meant touching all of them. The seam is small — the controller sends
// exactly two kinds of message that expect no answer, and asks two kinds of question — which is
// why the bus could be replaced at all.
//
// Two implementations live below. Both ship: steps 1 to 4 leave every node on AMQP
// ([ADR 0116](novox/hq)), so the controller keeps speaking it and the NATS one is selected at the
// rollout. That is also what makes them comparable — the same caller, the same arguments, and a
// conformance fixture holding both to one envelope.
type Bus interface {
	// PublishEvent announces something that happened, under the emitter's own name. 1:many, and
	// nobody is obliged to act (ADR 0041).
	PublishEvent(ctx context.Context, key, source, node string, body []byte) error

	// PublishDeclaration delivers one node what it should be. Addressed to that node alone: a
	// declaration is not an event, and replaying yesterday's is actively harmful
	// (design 29 §4, the *state* shape).
	PublishDeclaration(ctx context.Context, node string, body []byte) error
}

// --- AMQP, the bus the mesh runs on today -----------------------------------------------------

// OverAMQP is the bus as a channel.
type OverAMQP struct{ Channel *amqp.Channel }

func (b OverAMQP) PublishEvent(ctx context.Context, key, source, node string, body []byte) error {
	id, err := eventID()
	if err != nil {
		return err
	}
	return b.Channel.PublishWithContext(ctx, EventsExchange, key, false, false, amqp.Publishing{
		ContentType:  "application/json",
		DeliveryMode: amqp.Persistent,
		MessageId:    id,
		Timestamp:    time.Now().UTC(),
		Body:         body,
		Headers: amqp.Table{
			"x-event-id":   id,
			"x-source":     source,
			"x-node":       node,
			"x-time":       time.Now().UTC().Format(time.RFC3339),
			"content-type": "application/json",
		},
	})
}

func (b OverAMQP) PublishDeclaration(ctx context.Context, node string, body []byte) error {
	// To the queue directly rather than through an exchange: a declaration is for one node, and
	// routing it by name through a shared exchange would mean a binding per node that nothing
	// removes when a node is retired.
	return b.Channel.PublishWithContext(ctx, "", QueueFor(node), false, false, amqp.Publishing{
		ContentType:  "application/json",
		DeliveryMode: amqp.Persistent,
		Body:         body,
	})
}

// --- NATS, the bus being built ----------------------------------------------------------------

// OverNATS is the bus as a JetStream context.
type OverNATS struct{ JS nats.JetStreamContext }

// EventSubject is where a module's event lands. Derived from the emitter, never taken from the
// caller: a source that could differ from the subject is an envelope that can lie about its
// origin, and on NATS the account's permissions make the subject the authority (design 29 §2).
func EventSubject(source, key string) string {
	return "mesh.mod." + source + ".event." + key
}

// DeclareSubject is where one node's declaration lands. Last-per-subject on the NODES stream, so
// a node that was away gets exactly the current one and a replayed older one is refused by
// sequence — the wire-level answer to novox/hq issue 107.
func DeclareSubject(node string) string { return "mesh.node." + node + ".declare" }

func (b OverNATS) PublishEvent(ctx context.Context, key, source, node string, body []byte) error {
	id, err := eventID()
	if err != nil {
		return err
	}
	h := nats.Header{}
	h.Set("x-event-id", id)
	h.Set("x-source", source)
	h.Set("x-node", node)
	h.Set("x-time", time.Now().UTC().Format(time.RFC3339))
	h.Set("content-type", "application/json")

	// The id is also the publish's message id, so the server refuses a duplicate inside its
	// window. That narrows the window a consumer must deduplicate in; it does not remove the
	// requirement, because the window is finite (design 19, delivery).
	_, err = b.JS.PublishMsg(&nats.Msg{
		Subject: EventSubject(source, key),
		Header:  h,
		Data:    body,
	}, nats.MsgId(id), nats.Context(ctx))
	if err != nil {
		return fmt.Errorf("emitting %s: %w", key, err)
	}
	return nil
}

func (b OverNATS) PublishDeclaration(ctx context.Context, node string, body []byte) error {
	_, err := b.JS.Publish(DeclareSubject(node), body, nats.Context(ctx))
	if err != nil {
		return fmt.Errorf("declaring to %s: %w", node, err)
	}
	return nil
}