go-places/api/internal/events/events.go

// Package events publishes domain events so other services can react to them
// without being called synchronously.
package events

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

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

// Topic names the routing key an event is published under.
type Topic string

const (
	// TopicPlaceCreated fires once a place exists, before its pictures are
	// attached.
	TopicPlaceCreated Topic = "place.created"
	// TopicImageAnalyzed fires when a picture's badges are available.
	TopicImageAnalyzed Topic = "image.analyzed"
	// TopicImageLinked fires when a picture is linked to another taken at the
	// same place.
	TopicImageLinked Topic = "image.linked"
	// TopicCommentAdded fires when a comment is posted.
	TopicCommentAdded Topic = "comment.added"
)

// Event is the envelope every published message shares. Payload stays raw so
// consumers can evolve their own schema without the publisher changing.
type Event struct {
	// ID uniquely identifies this event, letting consumers deduplicate.
	ID string `json:"id"`
	// Topic is the routing key.
	Topic Topic `json:"topic"`
	// OccurredAt is when the event was produced.
	OccurredAt time.Time `json:"occurredAt"`
	// ActorID is the user who caused it, empty for system events.
	ActorID string `json:"actorId,omitempty"`
	// Payload is topic-specific JSON.
	Payload json.RawMessage `json:"payload"`
}

// Publisher sends events. Implementations must not block the request path for
// long; a slow broker should not stall an upload.
type Publisher interface {
	// Publish sends one event. Implementations should treat delivery as
	// best-effort and report failure to the caller rather than panicking.
	Publish(ctx context.Context, ev Event) error
	// Close releases resources.
	Close() error
}

// LogPublisher writes events to the structured log. It is the default so the
// service runs without a broker, and it keeps the event stream inspectable in
// development.
type LogPublisher struct {
	logger *slog.Logger
}

var _ Publisher = (*LogPublisher)(nil)

// NewLogPublisher returns a publisher that logs events.
func NewLogPublisher(logger *slog.Logger) *LogPublisher {
	if logger == nil {
		logger = slog.Default()
	}
	return &LogPublisher{logger: logger}
}

// Publish implements Publisher.
func (p *LogPublisher) Publish(ctx context.Context, ev Event) error {
	p.logger.InfoContext(ctx, "event published",
		"topic", string(ev.Topic),
		"eventId", ev.ID,
		"actorId", ev.ActorID,
		"payload", string(ev.Payload),
	)
	return nil
}

// Close implements Publisher.
func (p *LogPublisher) Close() error { return nil }

// AMQPPublisher publishes to a RabbitMQ topic exchange.
type AMQPPublisher struct {
	conn     *amqp.Connection
	channel  *amqp.Channel
	exchange string
	logger   *slog.Logger
	// confirms receives publisher acknowledgements, so a broker-side rejection
	// is noticed instead of being silently dropped.
	confirms <-chan amqp.Confirmation
}

var _ Publisher = (*AMQPPublisher)(nil)

// NewAMQP dials the broker, declares the exchange and enables confirms.
func NewAMQP(url, exchange string, logger *slog.Logger) (*AMQPPublisher, error) {
	if logger == nil {
		logger = slog.Default()
	}

	conn, err := amqp.Dial(url)
	if err != nil {
		return nil, fmt.Errorf("events: dial amqp: %w", err)
	}
	channel, err := conn.Channel()
	if err != nil {
		conn.Close()
		return nil, fmt.Errorf("events: open channel: %w", err)
	}

	// A topic exchange lets consumers bind to "image.*" or "place.created".
	if err := channel.ExchangeDeclare(exchange, "topic", true, false, false, false, nil); err != nil {
		channel.Close()
		conn.Close()
		return nil, fmt.Errorf("events: declare exchange %q: %w", exchange, err)
	}
	if err := channel.Confirm(false); err != nil {
		channel.Close()
		conn.Close()
		return nil, fmt.Errorf("events: enable confirms: %w", err)
	}

	return &AMQPPublisher{
		conn:     conn,
		channel:  channel,
		exchange: exchange,
		logger:   logger,
		confirms: channel.NotifyPublish(make(chan amqp.Confirmation, 1)),
	}, nil
}

// Publish implements Publisher. It waits briefly for the broker's
// acknowledgement so that a failed publish is reported to the caller.
func (p *AMQPPublisher) Publish(ctx context.Context, ev Event) error {
	body, err := json.Marshal(ev)
	if err != nil {
		return fmt.Errorf("events: marshal %s: %w", ev.Topic, err)
	}

	// Delivery mode 2 marks the message persistent.
	pub := amqp.Publishing{
		ContentType:  "application/json",
		DeliveryMode: amqp.Persistent,
		MessageId:    ev.ID,
		Timestamp:    ev.OccurredAt,
		Type:         string(ev.Topic),
		Body:         body,
	}

	if err := p.channel.PublishWithContext(ctx, p.exchange, string(ev.Topic), false, false, pub); err != nil {
		return fmt.Errorf("events: publish %s: %w", ev.Topic, err)
	}

	select {
	case confirm, ok := <-p.confirms:
		if !ok {
			return fmt.Errorf("events: confirmation channel closed for %s", ev.Topic)
		}
		if !confirm.Ack {
			return fmt.Errorf("events: broker rejected %s", ev.Topic)
		}
		return nil
	case <-ctx.Done():
		return ctx.Err()
	case <-time.After(confirmTimeout):
		return fmt.Errorf("events: no confirmation for %s within %s", ev.Topic, confirmTimeout)
	}
}

// confirmTimeout bounds the wait for a broker acknowledgement.
const confirmTimeout = 5 * time.Second

// Close implements Publisher.
func (p *AMQPPublisher) Close() error {
	var firstErr error
	if p.channel != nil {
		if err := p.channel.Close(); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	if p.conn != nil {
		if err := p.conn.Close(); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	return firstErr
}