eventsourcing

package module
v0.1.7 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 30, 2026 License: MPL-2.0 Imports: 12 Imported by: 0

README

TerraSkye Eventsourcing

eventsourcing is a generic, type-safe event sourcing framework for Go.
It provides the building blocks for event-driven architectures, including command handling, query handling, event buses, envelopes, and a flexible iterator for read models.

This library focuses on simplicity, modern Go patterns, and full generics support, making it easy to build event-sourced systems with strong type guarantees.

Features

  • Commands & Command Handlers – Strongly typed command routing.
  • Events & Event Handlers – Publish and consume events safely.
  • Event Bus – Supports multiple subscribers with typed handlers.
  • Queries & Query Handlers – Request data through a type-safe query bus.
  • Query Gateway – Simple façade to dispatch typed queries.
  • Generic Iterator – Lazy, paginated, or buffered read model iteration.
  • Revision Management – Built-in support for aggregate and stream revisions.
  • Metadata & Envelopes – Rich event metadata included by default.
  • Type-Safe Generics Everywhere – Commands, events, queries, handlers, results.

Installation

go get github.com/terraskye/eventsourcing

OpenTelemetry Instrumentation

The otel subpackage provides built-in observability for your event-sourced application using OpenTelemetry standards.

Why Instrumentation?

Event-sourced systems are inherently distributed and asynchronous. Without proper observability:

  • Command failures are hard to diagnose
  • Event handler latency goes unnoticed
  • Concurrency conflicts are invisible
  • Performance bottlenecks in the event store remain hidden

The otel package wraps your handlers and stores with tracing spans and metrics, giving you full visibility into your system's behavior without modifying business logic.

What's Instrumented
Component Spans Metrics
Command Handlers command.handle <Type> duration, in-flight, handled, failed, conflicts
Event Handlers events.handle <Type> duration, handled
Event Store EventStore.Save, EventStore.LoadStream, etc. duration, saves, loads, errors, events appended/loaded
How to Use

Wrap your handlers and stores with the telemetry decorators:

import "github.com/terraskye/eventsourcing/otel"

// Wrap a command handler
handler := otel.WithCommandTelemetry(myCommandHandler)

// Wrap an event handler
eventHandler := otel.WithEventTelemetry(myEventHandler)

// Wrap an event store
store := otel.WithEventStoreTelemetry(myEventStore)
Configuration Options

Customize span names and attributes using options:

// Static operation name
handler := otel.WithCommandTelemetry(myHandler,
    otel.WithOperation("order.create"),
)

// Add static attributes to all spans
handler := otel.WithCommandTelemetry(myHandler,
    otel.WithAttributes(
        attribute.String("service.name", "orders"),
        attribute.String("service.version", "1.0.0"),
    ),
)

// Dynamic operation name based on context
handler := otel.WithCommandTelemetry(myHandler,
    otel.WithOperationGetter(func(ctx context.Context, defaultOp string) string {
        if tenant := TenantFromContext(ctx); tenant != "" {
            return fmt.Sprintf("%s [%s]", defaultOp, tenant)
        }
        return defaultOp
    }),
)

// Extract dynamic attributes from context
handler := otel.WithCommandTelemetry(myHandler,
    otel.WithAttributeGetter(func(ctx context.Context) []attribute.KeyValue {
        return []attribute.KeyValue{
            attribute.String("tenant.id", TenantFromContext(ctx)),
            attribute.String("user.id", UserFromContext(ctx)),
        }
    }),
)
Available Metrics

Commands:

  • eventsourcing.commands.handled - total successful commands
  • eventsourcing.commands.failed - total failed commands
  • eventsourcing.commands.duration - histogram of handling time (ms)
  • eventsourcing.commands.in_flight - currently processing
  • eventsourcing.concurrency.conflicts - optimistic locking conflicts

Events:

  • eventsourcing.eventbus.handled - events processed by handlers
  • eventsourcing.eventbus.duration - handler execution time (ms)
  • eventsourcing.events.appended - events written to store
  • eventsourcing.events.loaded - events read from store

Event Store:

  • eventsourcing.eventstore.saves - save operations
  • eventsourcing.eventstore.duration - operation time (ms)
  • eventsourcing.eventstore.errors - failed operations
Semantic Attributes

All spans include semantic attributes following OpenTelemetry conventions:

  • eventsourcing.command.type - the command type name
  • eventsourcing.aggregate.id - target aggregate ID
  • eventsourcing.stream.id - event stream identifier
  • eventsourcing.stream.version - stream version after operation
  • eventsourcing.event.type - event type name
  • eventsourcing.event.id - unique event ID
  • eventsourcing.event.global_position - global ordering position
  • eventsourcing.event.stream_position - position within stream

Documentation

Index

Examples

Constants

View Source 
const (
	InstrumentationVersion = "v0.1.7"
)

InstrumentationVersion is the module version reported as the instrumentation version of this package's OpenTelemetry meter and tracer.

Variables

View Source 
var (
	// ErrStreamNotFound is returned by an [EventStore] when [StreamExists] is
	// required for a stream that does not exist.
	ErrStreamNotFound = errors.New("stream not found")
	// ErrStreamExists is returned by an [EventStore] when [NoStream] is
	// required for a stream that already exists.
	ErrStreamExists = errors.New("stream already exists")
	// ErrInvalidEventBatch is returned by [EventStore.Save] when the given
	// envelopes do not all share the same stream ID.
	ErrInvalidEventBatch = errors.New("invalid event batch")
	// ErrHandlerNotFound is returned by a [QueryBus] when no [QueryHandler]
	// is registered for a query's type.
	ErrHandlerNotFound = errors.New("handler not registered")
	// ErrInvalidRevision is returned by an [EventStore] when given a
	// [StreamState] implementation it does not support.
	ErrInvalidRevision = errors.New("invalid revision")
	// ErrHandlerNotRegistered is returned by a [CommandBus] when no
	// [CommandHandler] is registered for a command's type.
	ErrHandlerNotRegistered = errors.New("no handler registered for type")
	// ErrDuplicateHandler is returned or panicked with when registering a
	// second handler for a command or event type that already has one, by
	// [CommandBus.Register], [EventBus.Subscribe] implementations, and
	// [QueryBus] registration.
	ErrDuplicateHandler = errors.New("duplicate handler registered ")
	// ErrHandlerPanicked is joined into the error a [CommandBus] returns
	// when a [CommandHandler] panics instead of returning an error.
	ErrHandlerPanicked = errors.New("handler panicked when handling command")
	// ErrCommandBusClosed is wrapped into the error a [CommandBus] returns
	// when Dispatch is called after Stop.
	ErrCommandBusClosed = errors.New("command bus is closed")
	// ErrEventNotRegistered is returned by [NewEventByName] when no event
	// type is registered under the given name.
	ErrEventNotRegistered = errors.New("event not registered")
)
View Source 
var (

	// RegisterEventByType registers fn under the type name of the [Event] it
	// returns, i.e. fn().EventType(). Use this instead of [RegisterEvent]
	// when you need explicit control over the factory, for example to
	// inject constructor arguments. It panics if fn is nil, if fn() returns
	// nil, or if an event is already registered under that name.
	//
	// Example Usage:
	//   RegisterEventByType(func() Event { return &InventoryChanged{} })
	RegisterEventByType func(fn func() Event) = func(fn func() Event) {
		registerEventNameDefault(fn().EventType(), fn)
	}

	// RegisterEventByName registers fn under name, independently of the
	// event's own EventType() — useful when the name a store has events
	// persisted under no longer matches the current type name, for example
	// after a rename. As with [RegisterEventByType], it panics if fn is nil,
	// if fn() returns nil, or if name is already registered.
	//
	// Example Usage:
	//   RegisterEventByName("CustomEventName", func() Event { return &InventoryChanged{} })
	RegisterEventByName func(name string, fn func() Event) = func(name string, fn func() Event) {
		registerEventNameDefault(name, fn)
	}

	// NewEventByName returns a new instance of the [Event] registered under
	// name, or a non-nil [ErrEventNotRegistered] if no event is registered
	// under that name.
	//
	// Example Usage:
	//   ev, err := NewEventByName("InventoryChanged")
	NewEventByName func(name string) (Event, error) = newEventByNameDefault

	// EventNamesFor returns every name event's concrete type is registered
	// under, regardless of which [RegisterEvent]/[RegisterEventByType]/
	// [RegisterEventByName] call added each one.
	EventNamesFor func(event Event) []string = func(event Event) []string {
		registryMu.RLock()
		defer registryMu.RUnlock()

		return typeToNames[fmt.Sprintf("%T", event)]
	}
)

Functions

func AggregateIDFromContext 

func AggregateIDFromContext(ctx context.Context) string

AggregateIDFromContext returns the aggregate ID set by WithEnvelope, or "" if ctx carries none.

func CausationFromContext 

func CausationFromContext(ctx context.Context) string

CausationFromContext returns the causation ID set by WithCausation, or "" if ctx carries none.

func EventIDFromContext 

func EventIDFromContext(ctx context.Context) uuid.UUID

EventIDFromContext returns the event ID set by WithEnvelope, or uuid.Nil if ctx carries none.

func GlobalVersionFromContext 

func GlobalVersionFromContext(ctx context.Context) uint64

GlobalVersionFromContext returns the global version set by WithEnvelope, or 0 if ctx carries none.

func MetadataFromContext 

func MetadataFromContext(ctx context.Context) map[string]any

MetadataFromContext returns the metadata set by WithEnvelope, or nil if ctx carries none.

func NewBusinessRuleViolation added in v0.1.7 

func NewBusinessRuleViolation(err error) error

NewBusinessRuleViolation wraps err as an ErrBusinessRuleViolation. If err is nil, it returns nil.

NewCommandHandler already wraps whatever error its Decider returns in an ErrBusinessRuleViolation, so a Decider used with it should just return the plain error — calling this in a Decider would nest one violation inside another. Use it when signaling a business rule violation from a hand-rolled CommandHandler that doesn't go through NewCommandHandler's decide step, so callers (e.g. an OpenTelemetry middleware) can still recognize it as an expected, recoverable rejection via errors.As.

func OccurredAtFromContext 

func OccurredAtFromContext(ctx context.Context) time.Time

OccurredAtFromContext returns the occurred-at time set by WithEnvelope, or the zero time.Time if ctx carries none.

func Register 

func Register[C Command](b *CommandBus, handler CommandHandler[C])

Register installs handler as the handler for command type C on b. The registration key is derived from C with fmt.Sprintf("%T"), so there are no manual type strings to keep in sync. Register panics with ErrDuplicateHandler if a handler for C is already registered.

The middleware chain is applied here, at registration time, from the middlewares added via CommandBus.Use up to this point — so call Use first. Register is safe to call on a bus that is already dispatching, though wiring every handler during startup is the expected pattern.

Example: 

err := Register(bus, fooHandler)

func RegisterEvent 

func RegisterEvent[T any, PT eventPtr[T]](_ PT)

RegisterEvent registers the concrete event type T — inferred from the pointer passed in, whose value is otherwise discarded — under its default Event.EventType name. Each later NewEventByName call for that name returns a fresh new(T), so unlike a hand-written closure over a single instance, concurrent or repeated decodes never alias the same value. It panics if an event is already registered under that name.

Example Usage: 

RegisterEvent(&OrderCreated{})

func RegisterQueryHandler 

func RegisterQueryHandler[T Query, R any](bus *QueryBus, handler QueryHandler[T, R], opts ...HandlerOption)

RegisterQueryHandler registers a QueryHandler[T, R] on the bus. Use this when registering a type that explicitly implements the QueryHandler interface. For plain functions or method values, prefer RegisterQueryHandlerFunc.

Panics if a handler for the same query and result types is already registered.

Example Usage: 

RegisterQueryHandler(bus, myHandler)

func RegisterQueryHandlerFunc added in v0.1.6 

func RegisterQueryHandlerFunc[T Query, R any](bus *QueryBus, fn queryHandlerFunc[T, R], opts ...HandlerOption)

RegisterQueryHandlerFunc registers a plain function as a query handler. Type parameters are inferred from the function signature. Prefer this over RegisterQueryHandler when registering method values from a provider struct.

Panics if a handler for the same query and result types is already registered.

Example Usage: 

RegisterQueryHandlerFunc(bus, store.GetTask)
RegisterQueryHandlerFunc(bus, store.ListTasks)

func StreamIDFromContext 

func StreamIDFromContext(ctx context.Context) string

StreamIDFromContext returns the stream ID set by WithEnvelope, or "" if ctx carries none.

func TypeName 

func TypeName[T any](t T) string

TypeName returns t's concrete type name, without its own package path or any leading pointer asterisk. A generic type's name still includes its type argument(s), package-qualified, e.g. "Snapshot[eventsourcing.Cart]".

func VersionFromContext 

func VersionFromContext(ctx context.Context) uint64

VersionFromContext returns the stream version set by WithEnvelope, or 0 if ctx carries none.

func WithCausation 

func WithCausation(ctx context.Context, causation string) context.Context

WithCausation returns a copy of ctx carrying causation, the identifier of whatever caused the work now happening under ctx — for example, a command's type name — for handlers to attach to events or log entries.

func WithEnvelope 

func WithEnvelope(ctx context.Context, env *Envelope) context.Context

WithEnvelope returns a copy of ctx carrying env's stream ID, aggregate ID, event ID, version, global version, occurred-at time, and metadata, retrievable via the *FromContext functions below. It does not carry a causation ID; use WithCausation for that. The aggregate ID is "" if env.Event is nil.

Types

type Any 

type Any struct{}

Any expects nothing: append without checking the stream's current revision.

func (Any) ToRawInt64 

func (Any) ToRawInt64() int64

type AppendResult 

type AppendResult struct {
	// Successful reports whether the events were persisted.
	Successful bool
	// StreamID is the stream the events were saved to.
	StreamID string
	// NextExpectedVersion is the version the stream's next [Revision] should
	// use.
	NextExpectedVersion uint64
}

AppendResult describes the outcome of an EventStore.Save call.

type Command 

type Command interface {
	// AggregateID returns the ID of the entity this command targets, used to
	// locate that entity's state when handling the command.
	AggregateID() string
}

Command represents a user or system's intent to perform an action on a specific domain entity.

A Command models intent, not implementation: name it after what the user or system wants to achieve, in terms meaningful to the domain, rather than generic or mechanical terms like "Updated" or "Change". For example: 

Command Name      Intent
---------------- ----------------------------------------
ReserveSeat       Reserve a specific seat
CancelOrder       Cancel an existing order
ApproveUser       Approve a user registration
MarkInvoicePaid   Mark an invoice as paid
ShipOrder         Ship a customer order
CreateAccount     Create a new user account

Every Command must identify the entity it targets through Command.AggregateID, so a handler can locate that entity's state. A Command should be immutable after creation, representing a fixed intention at a point in time, and self-contained, carrying everything needed to handle it rather than relying on hidden state. For example: 

type ReserveSeat struct {
	ScreeningID string
	SeatNumber  string
	UserID      string
}

func (c ReserveSeat) AggregateID() string {
	return c.ScreeningID
}

type CommandBus 

type CommandBus struct {
	// contains filtered or unexported fields
}

CommandBus is an in-memory, type-safe command dispatcher. It keeps a registry of command type names to handlers, sharded queues of incoming commands, and the synchronization needed for concurrent [Dispatch] and Register calls. Commands for a given aggregate are always routed to the same shard and processed one at a time, [Stop] waits for in-flight dispatches to finish, and a handler panic is recovered rather than crashing the bus.

func NewCommandBus 

func NewCommandBus(bufferSize int, shardCount int) *CommandBus

NewCommandBus returns a CommandBus with shardCount shards, each with a worker goroutine already running and a queue buffering up to bufferSize commands. shardCount is clamped to at least 1.

Example: 

bus := NewCommandBus(100, 4)

func (*CommandBus) Dispatch 

func (b *CommandBus) Dispatch(ctx context.Context, cmd Command) (AppendResult, error)

Dispatch enqueues cmd for the handler registered for its type and blocks until that handler returns. Commands for the same aggregate are routed to the same shard, and each shard is drained by a single worker, so they are never handled concurrently. Dispatch itself is safe to call concurrently.

The returned AppendResult carries the outcome of the append. The error is non-nil if ctx is cancelled before the command is enqueued or before the result arrives, if no handler is registered for the command's type, if the handler returns an error or panics, or if the bus has been stopped. Once the bus is stopped the error wraps ErrCommandBusClosed, including for a call already parked waiting to enqueue.

func (*CommandBus) Stop 

func (b *CommandBus) Stop()

Stop shuts down the bus. It stops accepting new commands, lets the workers finish whatever is already queued, and waits for every in-flight CommandBus.Dispatch to return before it does.

A Dispatch racing Stop either completes normally or fails with an error wrapping ErrCommandBusClosed; it never panics. Stop is idempotent and safe to call concurrently, but the bus cannot be restarted afterwards.

func (*CommandBus) Use added in v0.1.6 

func (b *CommandBus) Use(middlewares ...CommandHandlerMiddleware)

Use appends middlewares to the chain applied to command handlers. The first one appended is the outermost wrapper and runs first on each dispatch.

Use must be called before Register: the chain is baked into each handler at registration time, so handlers already registered are not re-wrapped and a later Use call has no effect on them. Configure the full chain during startup wiring.

Example Usage: 

bus.Use(
    logging.CommandLogging(logger),
    otel.CommandTelemetry(),
)

type CommandHandler 

type CommandHandler[C Command] func(ctx context.Context, command C) (AppendResult, error)

CommandHandler handles commands of the concrete type C, which must implement Command. A handler carries out the business logic for a command — validating it, deciding what should happen, and persisting any resulting events — and is typically registered with a CommandBus via Register, which dispatches each command to the handler registered for its type.

It returns an AppendResult describing the outcome and a non-nil error if handling failed, for example due to a validation error, a business-rule violation, or a persistence failure. Implementations should treat the command as immutable, express state changes as events persisted to an EventStore rather than by mutating in-memory state, and return errors rather than panicking.

Example Usage: 

func HandleReserveSeat(ctx context.Context, cmd ReserveSeat) (AppendResult, error) {
    if seatAlreadyReserved(cmd.SeatNumber) {
        return AppendResult{Successful: false}, fmt.Errorf("seat already reserved")
    }
    // persist a SeatReserved event via an EventStore, then:
    return AppendResult{Successful: true, StreamID: cmd.AggregateID()}, nil
}

func NewCommandHandler 

func NewCommandHandler[T any, C Command](
	store EventStore,
	initialState InitialState[T],
	evolve Evolver[T],
	decide Decider[T, C],
	opts ...CommandHandlerOption,
) CommandHandler[C]

NewCommandHandler returns a generic command handler for any aggregate type.

The returned handler, on each call: loads command's stream from store (named by DefaultStreamNamer, or a StreamNamer set via WithStreamNamer) using EventStore.LoadStreamFrom; folds every loaded event into initialState with evolve; passes the resulting state and the command to decide to get the events to persist; wraps them in [Envelope]s, stamping each with a new UUID, the next sequential version, the current time, and any metadata from functions added via WithMetadataExtractor; and saves them with EventStore.Save.

The StreamState configured via WithStreamState (default Any) determines how a save conflict is handled. Any and StreamExists don't pin the stream to an exact version — Any asserts nothing, and StreamExists only asserts non-emptiness, once loading has confirmed the stream exists — so the handler saves against the revision it just loaded and, on conflict, retries the whole load-evolve-decide-save cycle according to the backoff.BackOff set via WithRetryStrategy (default: no retries). NoStream and a specific Revision pin the stream to an exact point (version 0, or N) that the caller explicitly asserted; a conflict there means that expectation was violated, so it is returned immediately rather than retried. A StreamExists load failure (the stream doesn't exist yet) is likewise always returned immediately, never retried — it's a fail-fast precondition, not a save conflict to converge on. Any other Save or load error is also returned directly, without retrying.

If decide returns no events, the handler returns a successful AppendResult without calling Save. If decide returns a non-nil error, the handler returns it wrapped in an ErrBusinessRuleViolation.

Example Usage: 

handler := NewCommandHandler(store, initialStateFunc, evolveFunc, decideFunc, WithStreamState(Any{}))
result, err := handler(ctx, myCommand)

type CommandHandlerMiddleware added in v0.1.6 

type CommandHandlerMiddleware func(next CommandHandler[Command]) CommandHandler[Command]

CommandHandlerMiddleware decorates a command handler on a CommandBus with a cross-cutting concern such as logging, telemetry, or rate limiting. It receives the next handler in the chain and returns a handler that wraps it; call next to pass the command along, or return without calling it to short-circuit.

The chain is baked into a handler once, by Register, from the middlewares added via CommandBus.Use up to that moment. Add all middleware during startup wiring, before registering any handler. The first middleware passed to Use is the outermost wrapper and runs first on each dispatch.

Example Usage: 

var rateLimiter eventsourcing.CommandHandlerMiddleware = func(
    next eventsourcing.CommandHandler[eventsourcing.Command],
) eventsourcing.CommandHandler[eventsourcing.Command] {
    return func(ctx context.Context, cmd eventsourcing.Command) (eventsourcing.AppendResult, error) {
        if !myLimiter.Allow() {
            return eventsourcing.AppendResult{}, errors.New("rate limit exceeded")
        }
        return next(ctx, cmd)
    }
}
bus.Use(rateLimiter)

type CommandHandlerOption 

type CommandHandlerOption func(configuration *handlerOptions)

CommandHandlerOption configures a CommandHandler built by NewCommandHandler.

func WithMetadataExtractor 

func WithMetadataExtractor(fn func(ctx context.Context) map[string]any) CommandHandlerOption

WithMetadataExtractor adds fn to the metadata functions a NewCommandHandler calls for every command, merging their results into each resulting Envelope's Metadata. Multiple extractors can be combined; they run in the order they were added, later ones overwriting keys set by earlier ones.

Usage: 

handler := NewCommandHandler(store, initialState, evolve, decide, WithMetadataExtractor(myMetadataFunc))

func WithRetryStrategy 

func WithRetryStrategy(strategy backoff.BackOff) CommandHandlerOption

WithRetryStrategy sets the backoff.BackOff a NewCommandHandler uses to retry its load-evolve-decide-save cycle after a save conflict, when the handler is configured (via WithStreamState) with Any (the default) or StreamExists — neither pins the stream to an exact version, so a save conflict can be resolved by retrying. It does not apply to a StreamExists load failure (the stream not existing yet), which is always returned immediately. Without this option, no retries are performed and the handler returns the conflict directly. It has no effect when WithStreamState is set to NoStream or a specific Revision — those pin an exact point, so a conflict there is always returned immediately.

Usage: 

handler := NewCommandHandler(store, initialState, evolve, decide, WithRetryStrategy(myBackoff))

func WithStreamNamer 

func WithStreamNamer(namer StreamNamer) CommandHandlerOption

WithStreamNamer overrides the StreamNamer a NewCommandHandler uses to name the stream it loads and saves to, in place of DefaultStreamNamer.

Usage: 

handler := NewCommandHandler(store, initialState, evolve, decide, WithStreamNamer(myStreamNamer))

func WithStreamState 

func WithStreamState(rev StreamState) CommandHandlerOption

WithStreamState sets the StreamState a NewCommandHandler expects the stream to be in when it saves events — Any (the default) to save against the revision the handler just loaded, or StreamExists to do the same after additionally requiring the stream to already exist; both retry on save conflict per WithRetryStrategy, since neither pins an exact version. (A StreamExists load failure — the stream not existing yet — is still a fail-fast precondition and is always returned immediately.) NoStream (stream must not exist) or a specific Revision (stream must be at exactly that version) pin the stream to an exact point the caller explicitly asserted; a conflict there is always returned immediately, never retried, since retrying would silently move past that point.

Usage: 

handler := NewCommandHandler(store, initialState, evolve, decide, WithStreamState(NoStream{}))

type Decider 

type Decider[T any, C Command] func(state T, cmd C) ([]Event, error)

Decider inspects state, as produced by an Evolver, against cmd and returns the events that should occur as a result, or a non-nil error if cmd violates a business rule or cannot be applied to state. It must not mutate state; state changes are expressed only through the returned events. An empty, nil slice means cmd produces no events, for example because it was idempotent or had no effect.

type Dispatcher 

type Dispatcher interface {
	Dispatch(ctx context.Context, cmd Command) (AppendResult, error)
}

Dispatcher has the same signature as CommandBus.Dispatch, for code that wants to depend on the dispatch behavior without depending on *CommandBus itself.

type Envelope 

type Envelope struct {
	// EventID uniquely identifies this occurrence of the event.
	EventID uuid.UUID
	// StreamID is the ID of the stream the event was appended to, typically
	// the aggregate's ID.
	StreamID string
	// Metadata carries caller-supplied, out-of-band information about the
	// event, such as correlation IDs or the acting user.
	Metadata map[string]any
	// Event is the wrapped domain event.
	Event Event
	// Version is the event's position within its own stream, starting at 1.
	Version uint64
	// GlobalVersion is the event's position across all streams, used to
	// resume a global subscription from a specific point.
	GlobalVersion uint64
	// OccurredAt is when the event was appended to the store.
	OccurredAt time.Time
}

Envelope wraps an Event with the metadata an EventStore or EventBus needs to persist, order, and route it.

type ErrBusinessRuleViolation 

type ErrBusinessRuleViolation struct {
	// contains filtered or unexported fields
}

ErrBusinessRuleViolation wraps an error returned when a Command violates a business rule. Use it to signal expected, recoverable domain-level rejections, as opposed to infrastructure or persistence errors. Its cause is unexported — construct one with NewBusinessRuleViolation and read the cause back with ErrBusinessRuleViolation.Cause or errors.Unwrap.

func (ErrBusinessRuleViolation) Cause 

func (e ErrBusinessRuleViolation) Cause() error

func (ErrBusinessRuleViolation) Error 

func (e ErrBusinessRuleViolation) Error() string

func (ErrBusinessRuleViolation) Unwrap 

func (e ErrBusinessRuleViolation) Unwrap() error

type ErrSkippedEvent 

type ErrSkippedEvent struct {
	Event Event
}

ErrSkippedEvent is returned when an EventHandler declines to process an Event because it is not the type the handler expects.

func (ErrSkippedEvent) Error 

func (e ErrSkippedEvent) Error() string

type Event 

type Event interface {
	// AggregateID returns the ID of the aggregate this event happened to.
	AggregateID() string
	// EventType returns the name under which this event's concrete type is
	// registered; see [RegisterEvent].
	EventType() string
}

Event is a domain event describing a change that has already happened to an aggregate.

type EventBus 

type EventBus interface {
	// Subscribe registers handler under name to receive events, optionally
	// narrowed by options. It returns an error if handler is nil, name is
	// already registered, or the subscription otherwise could not be added
	// (for example, for a networked implementation).
	Subscribe(ctx context.Context, name string, handler EventHandler, options ...SubscriberOption) error

	// Use adds middlewares to be applied to every handler registered via
	// Subscribe afterward. It must be called before Subscribe, since
	// middleware is applied at subscribe time.
	Use(middlewares ...EventHandlerMiddleware)

	// Errors returns a channel on which asynchronous handling errors are
	// delivered.
	Errors() <-chan error

	// Close shuts down the EventBus and waits for all handlers to finish
	// processing events already delivered to them.
	Close() error
}

EventBus lets [EventHandler]s subscribe to receive events as they occur, fanning each event out to every matching subscriber. How events are fed into the bus (typically as a side effect of an EventStore save) is left to the implementation.

type EventGroupProcessor 

type EventGroupProcessor struct {
	// contains filtered or unexported fields
}

EventGroupProcessor routes each incoming event to the EventHandler registered for its concrete type, typically one built with OnEvent.

func NewEventGroupProcessor 

func NewEventGroupProcessor(handlers ...EventHandler) *EventGroupProcessor

NewEventGroupProcessor builds an EventGroupProcessor from handlers, which must each implement an internal EventName() string method — as the handlers returned by OnEvent do — used as the routing key. It panics if a handler doesn't implement that method, or if two handlers report the same EventName().

Example Usage: 

p := &Projector{}
group := NewEventGroupProcessor(
    OnEvent(p.OnCartCreated),
    OnEvent(p.OnItemAdded),
)
group.Handle(ctx, CartCreated{ID: "t1"})
group.Handle(ctx, ItemAdded{ID: "c1"})

func (*EventGroupProcessor) Handle 

func (p *EventGroupProcessor) Handle(ctx context.Context, ev Event) error

Handle routes ev to the handler registered for its concrete type, or returns ErrSkippedEvent if none is registered.

func (*EventGroupProcessor) StreamFilter 

func (p *EventGroupProcessor) StreamFilter() []string

StreamFilter returns the sorted names of every event type this group has a handler for — useful, for example, as the filter passed to EventBus.Subscribe via a filtering SubscriberOption. For a handled type registered in the global event registry (see EventNamesFor) it includes every name that type is registered under, since one concrete event struct can be registered under several names (for example after a rename, via RegisterEventByName) and a subscriber needs to match all of them. For a handled type that isn't registered at all — registration is only needed by stores that rehydrate events by name, and has no bearing on what a group actually handles — it falls back to that type's own Event.EventType, so an unregistered handled type is never silently dropped from the filter.

type EventHandler 

type EventHandler interface {
	// Handle processes event within ctx.
	Handle(ctx context.Context, event Event) error
}

EventHandler processes events delivered by an EventBus or EventGroupProcessor.

func NewEventHandlerFunc 

func NewEventHandlerFunc(fn func(ctx context.Context, event Event) error) EventHandler

NewEventHandlerFunc returns fn as an EventHandler, for quickly wrapping a function without defining a separate type. fn is called for every event it is invoked with, unfiltered by type. The returned handler cannot be registered with an EventGroupProcessor — that requires each handler to report the single event type name it handles, which a handler for every event type has no one value for — so use OnEvent instead if you want a handler usable there, or if you only want to handle one concrete event type.

Example Usage: 

handler := NewEventHandlerFunc(func(ctx context.Context, ev Event) error {
    fmt.Println("Received event:", TypeName(ev))
    return nil
})
err := bus.Subscribe(ctx, "logger", handler)

func OnEvent 

func OnEvent[T Event](fn func(ctx context.Context, ev T) error) EventHandler

OnEvent returns fn as an EventHandler that only processes events of type T, returning ErrSkippedEvent for any other type. It is meant to be registered with an EventGroupProcessor, which uses T's type name to route only matching events to it.

Example Usage: 

handler := OnEvent(func(ctx context.Context, ev OrderCreated) error {
    fmt.Println("Order created:", ev.AggregateID())
    return nil
})
group := NewEventGroupProcessor(handler)
group.Handle(ctx, OrderCreated{ID: "123"})

type EventHandlerMiddleware added in v0.1.6 

type EventHandlerMiddleware func(next EventHandler) EventHandler

EventHandlerMiddleware decorates an event handler on an EventBus with a cross-cutting concern such as logging, telemetry, or error handling. It receives the next handler in the chain and returns a handler that wraps it; call next to pass the event along, or return without calling it to short-circuit. Middleware is applied when a handler is passed to EventBus.Subscribe, and the first middleware passed to EventBus.Use is the outermost wrapper and runs first for each event.

Example Usage: 

var myMiddleware eventsourcing.EventHandlerMiddleware = func(next eventsourcing.EventHandler) eventsourcing.EventHandler {
    return eventsourcing.NewEventHandlerFunc(func(ctx context.Context, event eventsourcing.Event) error {
        // before
        err := next.Handle(ctx, event)
        // after
        return err
    })
}

type EventStore 

type EventStore interface {
	// Save appends events to the stream identified by their common StreamID,
	// which must be the same across every element of events. revision states
	// the caller's expectation of the stream's current state — [Any] to
	// append unconditionally, [NoStream] to require the stream not already
	// exist, [StreamExists] to require that it does, or a specific [Revision]
	// to require an exact version — and Save returns a
	// [StreamRevisionConflictError] if that expectation does not hold.
	Save(ctx context.Context, events []Envelope, revision StreamState) (AppendResult, error)

	// LoadStream returns an iterator over every event in id's stream, in
	// ascending version order.
	LoadStream(ctx context.Context, id string) (*Iterator[*Envelope], error)

	// LoadStreamFrom returns an iterator over id's stream starting at
	// version, in ascending version order. version is typically a
	// [Revision]; [Any] starts from the beginning of the stream.
	LoadStreamFrom(ctx context.Context, id string, version StreamState) (*Iterator[*Envelope], error)

	// LoadFromAll returns an iterator over events across every stream,
	// starting at version. Whether version and the iteration order it
	// produces are globally consistent, or only consistent within each
	// stream, is implementation-specific; consult the implementation before
	// relying on cross-stream ordering.
	LoadFromAll(ctx context.Context, version StreamState) (*Iterator[*Envelope], error)

	// Close releases any resources held by the EventStore, such as network
	// connections or file handles. Implementations should make Close
	// idempotent. The EventStore must not be used after Close is called.
	Close() error
}

EventStore is an append-only store of [Envelope]s, grouped into per- aggregate streams, that allows an aggregate's state to be reconstructed by replaying its stream.

Implementations must store events for a given stream in the order they were saved and yield them in that same order — oldest first — from every Load* method. The Iterator values Load* methods return are lazy and should be consumed promptly; implementations make no guarantee about their reusability or thread-safety once iteration ends.

type EventStoreMiddleware added in v0.1.6 

type EventStoreMiddleware func(next EventStore) EventStore

EventStoreMiddleware decorates an EventStore, typically to intercept one or more of its methods while delegating the rest to next. Apply one by wrapping a store directly: store = mw(store). A common implementation embeds EventStore for the pass-through methods and overrides only the ones of interest.

Example Usage: 

var metered eventsourcing.EventStoreMiddleware = func(next eventsourcing.EventStore) eventsourcing.EventStore {
    return &meteredStore{next: next, counter: myCounter}
}
store = metered(baseStore)

type Evolver 

type Evolver[T any] func(currentState T, envelope *Envelope) T

Evolver applies a single historical event to currentState and returns the resulting aggregate state of type T. It must not mutate currentState; NewCommandHandler calls it once per event, in stream order, folding the result of each call into the next.

type GenericQueryGateway deprecated 

type GenericQueryGateway[T Query, R any] = QueryGateway[T, R]

GenericQueryGateway is a backwards-compatible alias for QueryGateway.

Deprecated: use QueryGateway directly.

type HandlerOption 

type HandlerOption func(*handlerSettings)

HandlerOption represents an optional configuration function that can modify handler behavior or metadata. Currently reserved for future extensions such as worker pools, timeouts, or rate limiting.

type InitialState 

type InitialState[T any] func() T

InitialState returns the zero-event aggregate state of type T, before any events have been evolved into it — for example, an empty struct or one with its fields set to their defaults.

type IterFunc 

type IterFunc[T any] func(ctx context.Context) (T, error)

IterFunc produces the next item of type T, returning io.EOF once iteration is complete.

type Iterator 

type Iterator[T any] struct {
	// contains filtered or unexported fields
}

Iterator is a pull-based iterator over items of type T, driven by an IterFunc supplied through NewIteratorFunc or NewSliceIterator. It underlies the event streams returned by an EventStore's Load* methods, but its next-function can equally paginate, stream, or buffer asynchronously — whatever the source needs.

Call Next repeatedly to advance the iterator, reading the current item with Value after each call that returns true; when Next returns false, call Err to distinguish a clean end of iteration (nil) from a failure. See [ExampleNewIteratorFunc].

func NewIteratorFunc 

func NewIteratorFunc[T any](nextFunc func(ctx context.Context) (T, error)) *Iterator[T]

NewIteratorFunc returns an Iterator driven by nextFunc, which must return io.EOF to signal the end of iteration and any other error to signal a failure.

Example 
package main

import (
	"context"
	"fmt"
	"io"

	cqrs "github.com/terraskye/eventsourcing"
)

func main() {
	items := []int{1, 2, 3, 4, 5}
	i := 0

	iter := cqrs.NewIteratorFunc(func(ctx context.Context) (int, error) {
		if i >= len(items) {
			return 0, io.EOF
		}
		val := items[i]
		i++
		return val, nil
	})

	for iter.Next(context.Background()) {
		fmt.Println(iter.Value())
	}
	if err := iter.Err(); err != nil {
		panic(err)
	}
}

Output:
1
2
3
4
5

func NewSliceIterator 

func NewSliceIterator[T any](slice []T) *Iterator[T]

NewSliceIterator returns an Iterator that yields the elements of slice in order.

func (*Iterator[T]) All 

func (it *Iterator[T]) All(ctx context.Context) ([]T, error)

All consumes the iterator by calling Next until it returns false, collecting each Value along the way, and returns the collected items along with the result of Err.

func (*Iterator[T]) Err 

func (it *Iterator[T]) Err() error

Err returns the error that ended iteration, or nil if Next has not yet returned false or iteration completed because the source was exhausted.

func (*Iterator[T]) Next 

func (it *Iterator[T]) Next(ctx context.Context) bool

Next advances the iterator and reports whether a new value is available. It returns false once the underlying IterFunc reports io.EOF or returns any other error; call Err afterward to tell the two apart.

func (*Iterator[T]) Value 

func (it *Iterator[T]) Value() T

Value returns the item read by the most recent call to Next, or the zero value of T if Next has not been called or has returned false.

type NoStream 

type NoStream struct{}

NoStream expects the stream not to exist yet.

func (NoStream) ToRawInt64 

func (NoStream) ToRawInt64() int64

type Query 

type Query interface {
	// ID returns an identifier for this query occurrence, for example for
	// correlation in logs or traces.
	ID() []byte
}

Query is implemented by any type that can be handled by a QueryHandler.

type QueryBus 

type QueryBus struct {
	// contains filtered or unexported fields
}

QueryBus is a central registry of query handlers, keyed by their query and result types, so that multiple query types can be registered on a single bus. Handlers are executed through a typed QueryGateway created with NewQueryGateway.

Example Usage: 

bus := NewQueryBus()
RegisterQueryHandlerFunc(bus, store.GetTask)
RegisterQueryHandlerFunc(bus, store.ListTasks)

func NewQueryBus 

func NewQueryBus() *QueryBus

NewQueryBus creates a new, empty QueryBus ready for handler registration.

func (*QueryBus) Use added in v0.1.6 

func (q *QueryBus) Use(middlewares ...QueryHandlerMiddleware)

Use appends middlewares to the chain applied to query handlers. The first one appended is the outermost wrapper and runs first on each query.

Use must be called before RegisterQueryHandler or RegisterQueryHandlerFunc: the chain is baked into each handler at registration time, so handlers already registered are not re-wrapped and a later Use call has no effect on them. Configure the full chain during startup wiring.

Example Usage: 

bus.Use(
    logging.QueryLogging(logger),
    otel.QueryTelemetry(),
)

func (*QueryBus) Validate 

func (q *QueryBus) Validate() error

Validate reports an error listing every query/result type pair that a QueryGateway was created for via NewQueryGateway but that has no registered handler. Call it during startup, after all gateways and handlers are wired up, to catch a missing registration before it can surface as a runtime error.

type QueryGateway added in v0.1.2 

type QueryGateway[T Query, R any] func(ctx context.Context, qry T) (R, error)

QueryGateway is a typed, callable facade over a QueryBus. Call it directly like a function to execute the handler registered for query type T and result type R. It also implements QueryHandler, so it can be passed to decorators such as WithQueryTelemetry or WithQueryLogging.

Example Usage: 

gateway := NewQueryGateway[MyQuery, *MyResult](bus)
result, err := gateway(ctx, MyQuery{ID: "42"})

func NewQueryGateway 

func NewQueryGateway[T Query, R any](bus *QueryBus) QueryGateway[T, R]

NewQueryGateway returns a QueryGateway for query type T and result type R, backed by bus. It registers the (T, R) pair on bus as a requestee, so that a later call to QueryBus.Validate fails if no handler for that pair is ever registered.

Example Usage: 

listGateway := NewQueryGateway[ListTasks, *TaskList](bus)
findGateway := NewQueryGateway[ListTasks, *Task](bus)

func (QueryGateway[T, R]) HandleQuery added in v0.1.2 

func (g QueryGateway[T, R]) HandleQuery(ctx context.Context, qry T) (R, error)

HandleQuery implements QueryHandler by calling g.

type QueryHandler 

type QueryHandler[T Query, R any] interface {
	HandleQuery(ctx context.Context, qry T) (R, error)
}

QueryHandler handles queries of the concrete type T, which must implement Query, and produces a result of type R — typically a read model or an Iterator over one. It enables generic, type-safe registration and execution of query logic through a QueryBus.

Example Usage: 

type MyQuery struct { ID string }
type MyResult struct { Value int }

handler := NewQueryHandlerFunc(func(ctx context.Context, q MyQuery) (*MyResult, error) {
    return &MyResult{Value: 123}, nil
})

var _ QueryHandler[MyQuery, *MyResult] = handler

func NewQueryHandlerFunc 

func NewQueryHandlerFunc[T Query, R any](fn func(ctx context.Context, qry T) (R, error)) QueryHandler[T, R]

NewQueryHandlerFunc returns fn as a QueryHandler.

Example Usage: 

handler := NewQueryHandlerFunc(func(ctx context.Context, q MyQuery) (*MyResult, error) {
    return &MyResult{Value: 42}, nil
})

type QueryHandlerMiddleware added in v0.1.6 

type QueryHandlerMiddleware func(next QueryGateway[Query, any]) QueryGateway[Query, any]

QueryHandlerMiddleware decorates a query handler on a QueryBus. It receives the next handler in the chain and returns a handler that wraps it; call next to pass the query along, or return without calling it to short-circuit.

The query arrives as a Query, so qry.ID is available directly; use fmt.Sprintf("%T", qry) for the concrete type name. The result arrives as an any holding the concrete result value.

The chain is baked into a handler once, by RegisterQueryHandler, from the middlewares added via QueryBus.Use up to that moment. Add all middleware during startup wiring, before registering any handler. The first middleware passed to Use is the outermost wrapper and runs first on each query.

Example Usage: 

var myMiddleware eventsourcing.QueryHandlerMiddleware = func(
    next eventsourcing.QueryGateway[eventsourcing.Query, any],
) eventsourcing.QueryGateway[eventsourcing.Query, any] {
    return func(ctx context.Context, qry eventsourcing.Query) (any, error) {
        // before
        result, err := next(ctx, qry)
        // after
        return result, err
    }
}
bus.Use(myMiddleware)

type Revision 

type Revision uint64

Revision expects the stream to be at exactly this version.

func (Revision) ToRawInt64 

func (r Revision) ToRawInt64() int64

type StreamExists 

type StreamExists struct{}

StreamExists expects the stream to already exist.

func (StreamExists) ToRawInt64 

func (StreamExists) ToRawInt64() int64

type StreamNamer 

type StreamNamer func(ctx context.Context, cmd Command) string

StreamNamer maps a Command to the name of the event stream it belongs to. 

var DefaultStreamNamer StreamNamer = func(ctx context.Context, cmd Command) string {
	return cmd.AggregateID()
}

DefaultStreamNamer is the StreamNamer used by command handlers when no custom namer is configured. By default it returns the command's Command.AggregateID as the stream name.

It can be overridden globally, for example to add tenant prefixes for multi-tenancy. Override it during program initialization, before any handler runs, to avoid data races on the global. 

// During startup:
DefaultStreamNamer = func(ctx context.Context, cmd Command) string {
	tenant, _ := ctx.Value(tenantKey).(string)
	return fmt.Sprintf("%s-orders-%s", tenant, cmd.AggregateID())
}

type StreamRevisionConflictError 

type StreamRevisionConflictError struct {
	Stream           string
	ExpectedRevision StreamState
	ActualRevision   StreamState
}

StreamRevisionConflictError is returned by an EventStore when a save is made against a Revision that no longer matches the stream's actual revision — an optimistic-concurrency conflict.

func (StreamRevisionConflictError) Error 

func (s StreamRevisionConflictError) Error() string

type StreamState 

type StreamState interface {
	// ToRawInt64 encodes the expectation as a store-specific raw value: a
	// non-negative [Revision] number, or one of the special negative
	// markers used by [Any] and [StreamExists].
	ToRawInt64() int64
}

StreamState expresses a caller's expectation of a stream's revision, for example to an EventStore.Save call via WithStreamState. Any, NoStream, StreamExists, and Revision are the implementations recognized by this package's EventStore implementations.

type SubscriberOption 

type SubscriberOption func(cfg any)

SubscriberOption configures a subscription registered via EventBus.Subscribe. Available options are implementation-specific; see, for example, the memory package's WithFilterEvents.

Source Files

View all Source files

Directories

Path Synopsis
eventbus
file
kurrentdb
memory
postgres
eventstore
file
kurrentdb
memory
postgres

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.