external_sink

eventsourced External Event Sink

Package external-sink integrates external event sources into the eventsourced framework by consuming events from AMQP and storing them in the event store.

Install

go get gitlab.com/unboundsoftware/eventsourced/external-sink

Prerequisites

• An AMQP broker (e.g., RabbitMQ)
• An eventsourced.EventStore implementation (e.g., pg)
• The goamqp library

Getting Started

Define external event types and create a sink:

import (
    external_sink "gitlab.com/unboundsoftware/eventsourced/external-sink"
    "gitlab.com/unboundsoftware/eventsourced/eventsourced"
)

// Define an external event type
type ExternalUserCreated struct {
    eventsourced.BaseEvent
    UserID string `json:"userId"`
    Name   string `json:"name"`
}

func (e *ExternalUserCreated) ExternalSource() string {
    return "user-service"
}

// Create the sink
sink, err := external_sink.New("my-service", eventStore,
    map[string]any{"User.Created": &ExternalUserCreated{}},
)
if err != nil {
    log.Fatal(err)
}

// Register AMQP setup with the connection
err = conn.Start(ctx, append(setups, sink.Setup(goamqp.WithDeadLetter())...)...)

Architecture

External events flow through these steps:

1. An external service publishes an event to AMQP
2. The sink's AMQP consumer receives the event
3. The event type is matched to a registered handler
4. An ExternalCommand wrapping the event is created
5. The command handler processes it, producing an eventsourced.ExternalEvent
6. The ExternalEvent is stored in the event store and published to the internal AMQP stream

Key Types

Sink

The Sink manages AMQP stream setup and event processing:

• New(serviceName, eventStore, types, opts...) - Creates a new sink. types maps each routing key to the external event type (a pointer implementing eventsourced.Event and eventsourced.ExternalSource) received on it; the event is re-published on the internal stream with the same key
• Setup(opts...) - Returns go-messaging-amqp setup functions for the internal stream publisher and a consumer bound to each registered key; opts apply to the consumer (e.g. WithDeadLetter())
• TypeMapper() - Resolves the registered keys to their types, for consumers of the internal stream
• StreamName() - Returns the AMQP stream name ("{serviceName}.internal")

ExternalAggregate

All external events are stored under a single aggregate with the fixed ID "external". This simplifies tracking without creating per-entity aggregates for external data.

ExternalCommand

Wraps an external event as a command. When processed, it produces an eventsourced.ExternalEvent[T] that stores both the original event and its source metadata.

CommandFunc

Maps an external event to an ExternalCommand:

type CommandFunc[T eventsourced.Event] func(source string, event T) ExternalCommand[T]

Event Type Registration

Event types must be passed as pointers implementing eventsourced.Event. JSON struct tags are validated at creation time to catch duplicate field names early. External event types should also implement the eventsourced.ExternalSource interface to identify their origin.

AMQP Stream

The sink creates an internal AMQP stream named "{serviceName}.internal" for republishing processed external events. This allows readviews and other consumers to subscribe to a unified event stream.