enkly-ai/pkg/eventbus/bus.go

package eventbus

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

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

// Bus is a small, reconnect-free wrapper over one amqp091 connection and
// channel. It declares ExchangeName as a durable topic exchange on
// connect so publishers and consumers never race over who creates it
// first.
//
// Deliberately minimal: no automatic reconnect/retry loop, no publisher
// confirms, no dead-letter exchange. Those are real production
// concerns, but adding them here would be infrastructure plumbing
// unrelated to the DDD restructuring this package exists to
// demonstrate — see the "Operational hardening" section of
// ARCHITECTURE.md for what a production rollout should add before this
// goes live.
type Bus struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	logger  *slog.Logger
}

// Connect dials url (an amqp:// URI) and declares the shared topology.
func Connect(url string, logger *slog.Logger) (*Bus, error) {
	conn, err := amqp.Dial(url)
	if err != nil {
		return nil, fmt.Errorf("dial rabbitmq: %w", err)
	}
	ch, err := conn.Channel()
	if err != nil {
		_ = conn.Close()
		return nil, fmt.Errorf("open channel: %w", err)
	}
	if err := ch.ExchangeDeclare(ExchangeName, "topic", true, false, false, false, nil); err != nil {
		_ = ch.Close()
		_ = conn.Close()
		return nil, fmt.Errorf("declare exchange: %w", err)
	}
	return &Bus{conn: conn, channel: ch, logger: logger}, nil
}

// Close releases the underlying channel and connection.
func (b *Bus) Close() error {
	if b == nil {
		return nil
	}
	if err := b.channel.Close(); err != nil {
		_ = b.conn.Close()
		return err
	}
	return b.conn.Close()
}

// Envelope is the JSON structure every event is published as. OccurredAt
// and RoutingKey are set by Publish; Payload is whatever event-specific
// struct the caller passes (see pkg/eventbus/events).
type Envelope struct {
	RoutingKey string          `json:"routing_key"`
	OccurredAt time.Time       `json:"occurred_at"`
	Payload    json.RawMessage `json:"payload"`
}

// Publish marshals payload and sends it to ExchangeName with the given
// routing key, as a persistent message (survives a broker restart).
// Fire-and-forget: the caller's own request/response flow (over gRPC)
// must never block on this succeeding — a publish failure is logged,
// not propagated, exactly like the monolith's own "log, don't fail the
// request" stance on e.g. a cleanup-loop error.
func (b *Bus) Publish(ctx context.Context, routingKey string, payload any) error {
	body, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("marshal event payload: %w", err)
	}
	env := Envelope{RoutingKey: routingKey, OccurredAt: time.Now().UTC(), Payload: body}
	envBody, err := json.Marshal(env)
	if err != nil {
		return fmt.Errorf("marshal envelope: %w", err)
	}
	return b.channel.PublishWithContext(ctx, ExchangeName, routingKey, false, false, amqp.Publishing{
		ContentType:  "application/json",
		DeliveryMode: amqp.Persistent,
		Timestamp:    time.Now(),
		Body:         envBody,
	})
}

// PublishBestEffort calls Publish and logs (rather than returns) any
// error — the shape every call site in this codebase actually wants,
// since a notification being late or lost must never fail the
// synchronous gRPC call that triggered it.
func (b *Bus) PublishBestEffort(ctx context.Context, routingKey string, payload any) {
	if err := b.Publish(ctx, routingKey, payload); err != nil {
		b.logger.Error("publish event", "routing_key", routingKey, "err", err)
	}
}

// Handler processes one decoded event. Returning an error nacks the
// delivery with requeue=true; a consumer that can't make progress
// should log and return nil instead of spinning forever on one poison
// message (the notification service does exactly this — see
// services/notification/internal/infrastructure/eventing).
type Handler func(ctx context.Context, env Envelope) error

// Consume declares binding.Queue durable, binds it to every routing key
// in binding.RoutingKeys, and runs handler for each delivery until ctx
// is cancelled. Blocks the calling goroutine.
func (b *Bus) Consume(ctx context.Context, binding QueueBinding, handler Handler) error {
	q, err := b.channel.QueueDeclare(binding.Queue, true, false, false, false, nil)
	if err != nil {
		return fmt.Errorf("declare queue %s: %w", binding.Queue, err)
	}
	for _, key := range binding.RoutingKeys {
		if err := b.channel.QueueBind(q.Name, key, ExchangeName, false, nil); err != nil {
			return fmt.Errorf("bind queue %s to %s: %w", binding.Queue, key, err)
		}
	}
	// Process one message at a time per consumer goroutine — the
	// notification service's send-an-email handler is I/O bound on
	// SMTP, not CPU bound, so this keeps ordering simple without
	// needing a worker pool for the volumes this is sized for.
	if err := b.channel.Qos(1, 0, false); err != nil {
		return fmt.Errorf("set qos: %w", err)
	}
	deliveries, err := b.channel.Consume(q.Name, "", false, false, false, false, nil)
	if err != nil {
		return fmt.Errorf("consume %s: %w", q.Name, err)
	}

	for {
		select {
		case <-ctx.Done():
			return nil
		case d, ok := <-deliveries:
			if !ok {
				return fmt.Errorf("delivery channel closed for queue %s", q.Name)
			}
			var env Envelope
			if err := json.Unmarshal(d.Body, &env); err != nil {
				b.logger.Error("decode event envelope", "queue", q.Name, "err", err)
				_ = d.Nack(false, false) // malformed — don't requeue
				continue
			}
			if err := handler(ctx, env); err != nil {
				b.logger.Error("handle event", "queue", q.Name, "routing_key", env.RoutingKey, "err", err)
				_ = d.Nack(false, true)
				continue
			}
			_ = d.Ack(false)
		}
	}
}