eventsourcing

package module
v0.1.2 Latest Latest
Warning

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

 Go to latest Published: Apr 29, 2026 License: MPL-2.0 Imports: 11 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

Constants

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

Variables

View Source 
var (
	ErrStreamNotFound       = errors.New("stream not found")
	ErrStreamExists         = errors.New("stream already exists")
	ErrInvalidEventBatch    = errors.New("invalid event batch")
	ErrHandlerNotFound      = errors.New("handler not registered")
	ErrInvalidRevision      = errors.New("invalid revision")
	ErrHandlerNotRegistered = errors.New("no handler registered for type")
	ErrDuplicateHandler     = errors.New("duplicate handler registered ")
	ErrHandlerPanicked      = errors.New("handler panicked when handling command")
	ErrCommandBusClosed     = errors.New("command bus is closed")
	ErrEventNotRegistered   = errors.New("event not registered")
)
View Source 
var (

	// RegisterEvent registers a new Event type using its default type name.
	//
	// It provides a reusable pattern for dynamically creating new event instances
	// by string name. Registration performs the following steps:
	//   1. Calls the provided factory function to obtain an instance of the event.
	//   2. Retrieves the type name using EventType().
	//   3. Registers the factory in the registry keyed by the type name.
	//
	// Parameters:
	//   - event: A factory function of type func() Event that returns a new instance
	//     of the event. The factory must not return nil.
	//
	// Panics:
	//   - If the factory function is nil.
	//   - If the factory returns nil.
	//   - If an event with the same type name is already registered.
	//
	// Example Usage:
	//   RegisterEvent(OrderCreated{})
	RegisterEvent func(event Event) = func(event Event) {
		RegisterEventByType(func() Event {
			return event
		})
	}
	// RegisterEventByType registers a new Event type using its default type name.
	//
	// It provides a reusable pattern for dynamically creating new event instances
	// by string name. Registration performs the following steps:
	//   1. Calls the provided factory function to obtain an instance of the event.
	//   2. Retrieves the type name using EventType().
	//   3. Registers the factory in the registry keyed by the type name.
	//
	// Parameters:
	//   - fn: A factory function of type func() Event that returns a new instance
	//     of the event. The factory must not return nil.
	//
	// Panics:
	//   - If the factory function is nil.
	//   - If the factory returns nil.
	//   - If an event with the same type name is already registered.
	//
	// Example Usage:
	//   RegisterEventByType(func() Event { return &InventoryChanged{} })
	RegisterEventByType func(fn func() Event) = func(fn func() Event) {
		registerEventNameDefault(fn().EventType(), fn)
	}

	// RegisterEventByName registers a new Event type under a custom name.
	//
	// This is similar to RegisterEventByType, but allows using a name
	// that is independent of EventType(). The provided factory function
	// must return a new instance of the event type each time it is called.
	//
	// Parameters:
	//   - name: The unique name to register the event under.
	//   - fn: Factory function of type func() Event that returns a new instance.
	//
	// Panics:
	//   - If fn is nil.
	//   - If fn returns nil.
	//   - If the 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 creates a new instance of a registered Event by its name.
	//
	// This function allows dynamic instantiation of events using their string name.
	// It performs the following steps:
	//   1. Looks up the factory function in the registry.
	//   2. Calls the factory to create a new instance.
	//
	// Parameters:
	//   - name: The name of the event to create.
	//
	// Returns:
	//   - Event: A new instance of the registered event.
	//   - error: Non-nil if the event name is not registered or the factory
	//     returned nil.
	//
	// Example Usage:
	//   ev, err := NewEventByName("InventoryChanged")
	NewEventByName func(name string) (Event, error) = newEventByNameDefault

	// EventNamesFor returns the registered names for a given Event type.
	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 AggregateID or "" if not present

func CausationFromContext 

func CausationFromContext(ctx context.Context) string

CausationFromContext returns Metadata or nil if not present

func EventIDFromContext 

func EventIDFromContext(ctx context.Context) uuid.UUID

EventIDFromContext returns the EventID or uuid.Nil if not present

func GlobalVersionFromContext 

func GlobalVersionFromContext(ctx context.Context) uint64

VersionFromContext returns the Version or 0 if not present

func MetadataFromContext 

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

MetadataFromContext returns Metadata or nil if not present

func OccurredAtFromContext 

func OccurredAtFromContext(ctx context.Context) time.Time

OccurredAtFromContext returns OccurredAt or zero time if not present

func Register 

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

Register adds a new typed command handler to the bus.

Parameters:

  • b: pointer to the CommandBus
  • handler: a generic CommandHandler[Command] function for a specific command type C

Notes:

  • Derives the command type name automatically using fmt.Sprintf("%T") to avoid manual registration strings.
  • Panics if a handler is already registered for the same command type.

Example: 

err := Register(bus, fooHandler)

func RegisterQueryHandler 

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

RegisterQueryHandler registers a QueryHandler for a specific query and result type on the provided QueryBus.

This function generates a unique key from the types of T and R, stores the handler in the bus, and applies any optional configuration.

Type Parameters:

  • T: The query type implementing query.Query.
  • R: The result type (ReadModel or Iterator).

Parameters:

  • bus: The QueryBus instance where the handler should be registered.
  • handler: The QueryHandler to register.
  • opts: Optional HandlerOption values for future customization.

Behavior Details:

  • The key for storage is generated via fmt.Sprintf("%T|%T").
  • Currently, handler settings are collected but not persisted.

Example Usage: 

bus := NewQueryBus()
RegisterQueryHandler[MyQuery, *MyResult](bus, NewQueryHandlerFunc(func(ctx context.Context, q MyQuery) (*MyResult, error) {
    return &MyResult{Value: 42}, nil
}))

Generic helper function

func StreamIDFromContext 

func StreamIDFromContext(ctx context.Context) string

StreamIDFromContext returns the StreamID or "" if not present

func TypeName 

func TypeName[T any](t T) string

TypeName returns the struct name without the package path

func VersionFromContext 

func VersionFromContext(ctx context.Context) uint64

VersionFromContext returns the Version or 0 if not present

func WithCausation 

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

func WithEnvelope 

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

WithEnvelope adds the context of the Event to the context

Types

type Any 

type Any struct{}

Any means append without checking current revision.

func (Any) ToRawInt64 

func (Any) ToRawInt64() int64

type AppendResult 

type AppendResult struct {
	Successful          bool
	StreamID            string
	NextExpectedVersion uint64
}

AppendResult describes the outcome of an append operation.

type Command 

type Command interface {
	// AggregateID returns the ID of the entity this command targets.
	// This ID is used to locate the entity in the system for processing the command.
	AggregateID() string
}

Command represents a user or system intention to perform an action on a specific domain entity.

A Command models **intent**, not implementation. It should describe **what the user or system wants to achieve**, in terms that are meaningful to the business or domain experts.

Key guidelines for designing Commands:

1. **Name by intent, not technical detail**:

  • The name should clearly describe the intention or purpose of the action.

  • Avoid generic or mechanical terms like "Updated" or "Change".

    Example command names and their intent:

Example command names and their intent: 

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

2. **Aggregate target**:

  • Every Command must identify the entity it targets through `AggregateID()` (or a similar method in your domain model).
  • This allows the handler to locate the relevant state.

3. **Immutability**:

  • Commands should be immutable after creation. They represent a fixed intention at a point in time.

4. **Self-contained**:

  • Include all information necessary for the Command to be handled correctly.
  • Do not rely on hidden state or external assumptions.

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 internal, in-memory, type-safe command dispatcher. It maintains a mapping of command type names to their handlers, a queues for incoming commands, and synchronization mechanisms for safe concurrent access.

The CommandBus supports:

  • Enqueuing commands for asynchronous processing
  • Typed command registration using generics
  • Safe shutdown that waits for in-flight commands to complete
  • Panic recovery in handlers to prevent the bus from crashing

func NewCommandBus 

func NewCommandBus(bufferSize int, shardCount int) *CommandBus

NewCommandBus creates a new instance of CommandBus with a buffered queues.

Parameters:

  • bufferSize: the size of the internal queues for enqueued commands.

Returns:

  • pointer to a newly initialized CommandBus. The internal processing goroutine is started automatically.

Example: 

bus := NewCommandBus(100)

func (*CommandBus) Dispatch 

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

Dispatch enqueues a command for processing by the registered handler and waits for the result. It is safe to call concurrently.

Parameters:

  • ctx: the context for cancellation or timeout
  • cmd: the command to dispatch

Returns:

  • AppendResult: indicates success/failure of command processing
  • error: non-nil if the dispatch failed due to context cancellation or processing error

Notes:

  • Returns an error immediately if the bus has been stopped.
  • Waits for the handler to complete and sends the result back via a response channel.

func (*CommandBus) Stop 

func (b *CommandBus) Stop()

Stop shuts down the CommandBus safely.

Behavior:

  • Stops accepting new commands.
  • Closes the internal queues channel.
  • Waits for all in-flight commands to finish before returning.

Example: 

bus.Stop()

type CommandHandler 

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

CommandHandler defines a function type for handling commands of a specific type.

C represents the concrete command type implementing the Command interface.

A CommandHandler is responsible for implementing the business logic associated with a command. This typically includes validation, orchestration, and producing side effects, such as persisting events to an EventStore or triggering other operations.

Handlers of this type are generally registered with a CommandBus, which ensures that commands are dispatched to the correct handler based on their type.

Parameters:

  • ctx: The context for controlling cancellation, deadlines, and carrying request-scoped values.
  • command: The command of type C, representing the intent to perform a domain action.

Returns:

  • AppendResult: Represents the result of handling the command, including success status, the next expected version of the aggregate, and any events that were persisted.
  • error: Non-nil if the command handling failed, e.g., due to validation errors, business rule violations, or persistence failures.

Notes:

  • Implementations should treat the command as immutable.
  • Any domain state changes should be expressed via events (AppendResult.Events) rather than directly mutating state.
  • Handlers should not panic; all errors should be returned via the error return value.

Example Usage: 

func HandleReserveSeat(ctx context.Context, cmd ReserveSeat) (AppendResult, error) {
    if seatAlreadyReserved(cmd.SeatNumber) {
        return AppendResult{Successful: false}, fmt.Errorf("seat already reserved")
    }
    events := []Event{SeatReserved{SeatNumber: cmd.SeatNumber, UserID: cmd.UserID}}
    return AppendResult{Successful: true, Events: events}, 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.

It provides a reusable pattern for handling commands in an event-sourced system by performing the following steps:

  1. Load the event history for the aggregate (using LoadStreamFrom).
  2. Evolve the current state based on the event history.
  3. Decide which new events should occur based on the command and current state.
  4. Wrap the decided events in envelopes, assigning version numbers and metadata.
  5. Persist the envelopes to the EventStore, respecting the configured revision and concurrency rules.

Parameters:

  • store: The EventStore used to load and persist events.
  • initialState: a function of type InitialState[T] that created the initial state for the command.
  • evolve: A function of type Evolver[T] that reconstructs aggregate state from a sequence of events.
  • decide: A function of type Decider[T, C] that produces events based on the current state and command.
  • opts: Optional CommandHandlerOption values for customizing behavior, such as:
  • StreamState: The expected stream revision (default is Any).
  • RetryAttempts: Number of retries on version conflicts (default 0).

Returns:

  • A function that takes a context and a command of type C, and returns:
  • AppendResult: Contains information about the persistence result, including success and the next expected version.
  • error: Non-nil if the command failed, either due to a business rule violation, persistence error, or concurrency conflict.

Behavior Details:

  • The command’s AggregateID() is used to identify the target stream in the EventStore.
  • The SeqWithSideEffect wrapper tracks the last version while evolving state.
  • If the configured StreamState is Revision, it is updated to the latest version before saving to ensure optimistic concurrency control.
  • If the decide function returns no events, the handler returns a successful result without persisting.
  • Each event is wrapped in an Envelope with a new UUID, metadata map, version, and timestamp.
  • Errors during loading, evolving, deciding, or saving are propagated with context using errors.Wrap.

Example Usage: 

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

type CommandHandlerOption 

type CommandHandlerOption func(configuration *handlerOptions)

CommandHandlerOption defines a function type that modifies handlerOptions. These options are applied when constructing a NewCommandHandler to customize behavior.

func WithMetadataExtractor 

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

WithMetadataExtractor adds a metadata function to a NewCommandHandler.

Each metadata function is called for every command handling execution and can inject additional key-value pairs into the event envelopes. Multiple metadata extractors can be combined; they are applied in order of registration.

Usage: 

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

func WithRetryStrategy 

func WithRetryStrategy(strategy backoff.BackOff) CommandHandlerOption

WithRetryStrategy sets the retry strategy for a NewCommandHandler.

The BackOff strategy controls how many times and with what delay the handler retries saving events in case of concurrency conflicts or transient errors.

Usage: 

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

func WithStreamNamer 

func WithStreamNamer(namer StreamNamer) CommandHandlerOption

WithStreamNamer sets a custom stream naming function on a NewCommandHandler.

The StreamNamer is called for every command handling execution and determines the stream name used to load and persist events. This allows customization beyond the default behavior of using the command's AggregateID as the stream name.

Usage: 

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

func WithStreamState 

func WithStreamState(rev StreamState) CommandHandlerOption

WithStreamState sets the expected stream revision for a NewCommandHandler.

The StreamState controls the concurrency check when persisting events. For example:

  • Any{}: no version check (default)
  • NoStream{}: ensures the stream does not exist
  • StreamExists{}: ensures the stream exists
  • Revision{N}: expects the stream to be at version N

Usage: 

handler := NewCommandHandler(store, initialState, evolve, decide, WithRevision(NoStream))

type Decider 

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

Decider determines which events should occur based on the current state and a command.

T represents the aggregate state type. C represents the command type.

Parameters:

  • state: The current aggregate state as returned by the Evolver.
  • cmd: The command to handle, containing the intent to change state.

Returns:

  • A slice of Event representing the events that should be applied to the aggregate.
  • An error, which should be non-nil if the command violates business rules or cannot be applied to the current state.

Notes:

  • The Decider should not mutate the input state directly; it should produce events that, when applied via the Evolver, will update the state accordingly.
  • Returning an empty slice indicates that the command produces no events (e.g., it was idempotent or had no effect).

type Dispatcher 

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

type Envelope 

type Envelope struct {
	EventID       uuid.UUID
	StreamID      string
	Metadata      map[string]any
	Event         Event
	Version       uint64
	GlobalVersion uint64
	OccurredAt    time.Time
}

type ErrBusinessRuleViolation 

type ErrBusinessRuleViolation struct {
	Err error
}

ErrBusinessRuleViolation is returned when a command violates a business rule. Use this to signal domain-level rejections that are expected and recoverable, as opposed to infrastructure or persistence errors.

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 a handler cannot handle the event type.

func (ErrSkippedEvent) Error 

func (e ErrSkippedEvent) Error() string

type Event 

type Event interface {
	AggregateID() string
	EventType() string
}

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

type EventBus 

type EventBus interface {
	// Subscribe adds a handler for an event. Returns an error if either the
	// matcher or handler is nil, the handler is already added or there was some
	// other problem adding the handler (for networked handlers for example).
	Subscribe(ctx context.Context, name string, handler EventHandler, options ...SubscriberOption) error

	// Errors returns an error channel where async handling errors are sent.
	Errors() <-chan error

	// Close closes the EventBus and waits for all handlers to finish.
	Close() error
}

EventBus is an EventHandler that distributes published events to all matching handlers that are registered, but only one of each type will handle the event.

type EventGroupProcessor 

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

EventGroupProcessor is a collection of typed event handlers. It routes incoming events to the correct handler based on event type.

func NewEventGroupProcessor 

func NewEventGroupProcessor(handlers ...EventHandler) *EventGroupProcessor

NewEventGroupProcessor creates a group of typed event handlers.

It provides a reusable pattern for routing events to the correct handler by performing the following steps:

  1. Accepts a list of typed EventHandler instances (created via OnEvent).
  2. Validates that all handlers implement EventName().
  3. Builds an internal map from EventName() to EventHandler for fast routing.
  4. Panics if duplicate handlers are provided for the same event type.

Parameters:

  • handlers: A variadic list of typed EventHandler instances.

Returns:

  • *eventGroupProcessor: a processor that routes events to the appropriate handler.

Behavior Details:

  • Events are dispatched based on the EventName() returned by the handler.
  • If no handler exists for an event, Handle returns ErrSkippedEvent.
  • StreamFilter() returns the sorted list of event names handled by the group.

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 the given event to the correct typed handler. Returns ErrSkippedEvent if no handler exists for the event type.

func (*EventGroupProcessor) StreamFilter 

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

StreamFilter returns a sorted list of all event names handled by this group. Useful for subscribing to streams or listing registered handlers.

type EventHandler 

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

EventHandler represents a generic event handler that can handle an Event.

func NewEventHandlerFunc 

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

NewEventHandlerFunc creates an EventHandler from a plain function.

This is a helper for quickly creating an EventHandler without defining a separate struct. It wraps the provided function and implements the EventHandler interface, allowing it to be used wherever an EventHandler is required, such as in an EventHandlerGroup.

Parameters:

  • fn: A function with signature func(ctx context.Context, ev Event) error. This function will be called whenever the EventHandler receives an event.

Returns:

  • EventHandler: an implementation of the EventHandler interface that delegates event handling to the provided function.

Behavior Details:

  • The provided function is called for every event passed to the EventHandler.
  • There is no type-checking or filtering: the handler will receive all events that it is invoked with. If you need type safety, use OnEvent[T] instead.
  • Any error returned by the function is propagated directly to the caller.

Example Usage: 

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

// Can be used in an EventHandlerGroup
group := NewEventGroupProcessor(handler)
group.Handle(ctx, MyEvent{ID: "123"})

func OnEvent 

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

OnEvent creates a strongly-typed EventHandler for a specific event type.

It provides a reusable pattern for handling events in a type-safe manner by performing the following steps:

  1. Wraps a user-provided function `fn(ctx, ev T)` into a typed handler.
  2. Associates the handler with the type name of T for internal routing.
  3. Returns an EventHandler that can be registered in an eventGroupProcessor.

Parameters:

  • fn: A function with signature func(ctx context.Context, ev T) error, where T implements the Event interface.

Returns:

  • EventHandler: a strongly-typed handler for events of type T.

Behavior Details:

  • When called via eventGroupProcessor.Handle, the handler will only receive events of type T. If a different event type is passed, it returns ErrSkippedEvent.
  • EventName() internally derives the type name of T using TypeName[T].

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 EventStore 

type EventStore interface {
	// Save appends all events in the given slice to the event stream for a specific aggregate.
	//
	// Parameters:
	//   - ctx: Request-scoped context for cancellation and tracing.
	//   - events: A slice of Envelope values to append. Each envelope should have
	//     the aggregate ID and version set consistently.
	//   - revision: The expected stream state or concurrency requirement. This can
	//     be one of:
	//       - Any: always append, do not check for conflicts.
	//       - NoStream: stream must not exist; fail if it does.
	//       - StreamExists: stream must exist; fail if it does not.
	//
	// Errors:
	//   - ErrConcurrency if the originalVersion does not match.
	//   - Any store-specific persistence error.
	Save(ctx context.Context, events []Envelope, revision StreamState) (AppendResult, error)

	// LoadStream loads all events for the given aggregate ID from version 0 onward.
	//
	// The returned iterator yields events in ascending version order.
	// Iteration stops if:
	//   - The iterator function returns false (consumer stops early).
	//   - The context is canceled.
	//
	// Returns:
	//   - iter.Seq[*Envelope]: Lazy iterator over events.
	//   - error: Non-nil if the store could not read events.
	LoadStream(ctx context.Context, id string) (*Iterator[*Envelope], error)

	// LoadStreamFrom loads all events for the given aggregate ID starting at the specified version.
	//
	// Parameters:
	//   - ctx: Request-scoped context for cancellation and tracing.
	//   - id: Aggregate identifier.
	//   - version: Zero-based version index from which to start iteration.
	//
	// Returns:
	//   - EnvelopeIterator: Lazy iterator over events from
	LoadStreamFrom(ctx context.Context, id string, version StreamState) (*Iterator[*Envelope], error)

	// LoadFromAll loads all events from all aggregates starting at the specified version index.
	//
	// The definition of "version" here is store-specific; it may represent:
	//   - A global monotonically increasing sequence number across all aggregates.
	//   - A local version within each aggregate (in which case, events are yielded
	//     from each aggregate starting at that version).
	//
	// Events should be yielded in chronological order as stored by the backend.
	// Consumers should not assume global ordering unless explicitly documented by
	// the implementation.
	LoadFromAll(ctx context.Context, version StreamState) (*Iterator[*Envelope], error)
	// Close releases any resources held by the EventStore, such as network
	// connections or file handles. After Close is called, the EventStore should
	// not be used.
	//
	// Implementations should make Close idempotent.
	Close() error
}

EventStore defines the contract for an append-only event store used in event-sourced systems. An EventStore persists events associated with a given aggregate ID in sequential order, allowing for full reconstruction of aggregate state at any point in time.

Implementations must guarantee:

  • Events for a given aggregate are stored in order.
  • Concurrency control based on the aggregate's expected version.
  • Iteration order from all Load* methods is deterministic (oldest → newest).

The returned iter.Seq values are lazy iterators over the stored events. They should be consumed immediately; no assumptions should be made about reusability or thread-safety after iteration completes.

type Evolver 

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

Evolver evolves the given state into a new state with the event applied.

T represents the aggregate state type.

Parameters:

  • currentState: The current aggregate state.
  • envelope: A Envelope object representing an historical event of an aggregate.

Returns:

  • The reconstructed aggregate state of type T.

Notes:

  • The Evolver is responsible for applying the event to the current state, producing the latest state.

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 an initial state of type T

T represents the aggregate state type.

Returns:go

  • The initial aggregate state of type T.

Notes:

  • The InitialState is responsible for returning an initial state

type IterFunc 

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

type Iterator 

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

Iterator is a generic, type-safe iterator over items of type T.

It abstracts the iteration logic, allowing for multiple iteration strategies such as paginated fetching, streaming, or pre-filling a buffer asynchronously.

The iterator provides the following API:

  • Next(ctx): advances the iterator and reports whether a next value exists.
  • Value(): retrieves the current value after Next() returns true.
  • Err(): retrieves any error encountered during iteration.
  • All(ctx): consumes the iterator fully and returns all items as a slice.

Type Parameters:

  • T: The item type returned by the iterator. Can be a pointer, struct, or primitive.

Example Usage: 

items := []int{1, 2, 3, 4, 5}
i := 0
iter := 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)
}

func NewIteratorFunc 

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

NewIteratorFunc constructs a new Iterator[T] using the provided nextFunc.

Parameters:

  • nextFunc: function that produces the next item. Must return: (T, nil) for valid items (any, io.EOF) to signal end of iteration (any, error) to signal a failure

Returns:

  • *Iterator[T]: a new iterator ready for consumption via Next()/All()

Example: 

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

func NewSliceIterator 

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

NewSliceIterator constructs a new Iterator[T] using the provided slice T.

Parameters:

  • slice: []T

Returns:

  • *Iterator[T]: a new iterator ready for consumption via Next()/All()

Example: 

iter := NewSliceIterator([]int{1,2,3,4,5})

func (*Iterator[T]) All 

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

All consumes the iterator and returns all remaining items in a slice.

Returns:

  • []T: all items produced by the iterator
  • error: the first non-EOF error encountered, or nil if iteration completed normally.

Behavior:

  • Repeatedly calls Next() until it returns false.
  • Collects all items via Value().
  • Returns any error encountered via Err().

func (*Iterator[T]) Err 

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

Err returns the last error encountered during iteration.

Returns:

  • error: the error returned by nextFunc, if any. Returns nil if iteration completed normally.

func (*Iterator[T]) Next 

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

Next advances the iterator to the next value.

Returns:

  • bool: true if a new value is available; false if iteration is complete or an error occurred.

Behavior:

  • Calls nextFunc to retrieve the next item.
  • If nextFunc returns io.EOF, marks iteration as done and returns false.
  • If nextFunc returns a non-nil error, stores it and returns false.
  • Otherwise, stores the item in current and returns true.

Example: 

if iter.Next(ctx) {
    item := iter.Value()
}

func (*Iterator[T]) Value 

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

Value returns the current item in the iteration.

Returns:

  • T: the current item, or the zero value if Next() has not been called or iteration has completed.

Usage: 

item := iter.Value()

type NoStream 

type NoStream struct{}

NoStream means the stream should not exist yet.

func (NoStream) ToRawInt64 

func (NoStream) ToRawInt64() int64

type Query 

type Query interface {
	ID() []byte
}

Query is the interface that must be implemented by any type to be considered a query.

type QueryBus 

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

QueryBus acts as a central registry for query handlers. It stores handlers keyed by their query and result types, allowing multiple query types to be registered in a single bus.

Handlers can later be executed via a typed GenericQueryGateway.

Example Usage: 

bus := NewQueryBus()
RegisterQueryHandler[MyQuery, *MyResult](bus, NewQueryHandlerFunc(func(ctx context.Context, q MyQuery) (*MyResult, error) {
    return &MyResult{Value: 42}, nil
}))

func NewQueryBus 

func NewQueryBus() *QueryBus

NewQueryBus creates a new QueryBus instance.

Returns:

  • *QueryBus: A new, empty bus ready for handler registration.

func (QueryBus) Validate 

func (q QueryBus) Validate() 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 QueryBus. Call it directly like a function to execute the registered handler for query type T. It also implements QueryHandler[T, R], so it can be passed to decorators such as WithQueryTelemetry or WithQueryLogging.

Type Parameters:

  • T: The query type implementing Query.
  • R: The result type.

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 creates a QueryGateway for a specific query and result type backed by a QueryBus. Creating a gateway registers the (T, R) pair as a requestee, which is checked by bus.Validate() at startup.

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[T, R].

type QueryHandler 

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

QueryHandler represents a handler for a specific query type T and produces a result of type R. This interface allows generic, type-safe registration and execution of query logic.

Type Parameters:

  • T: The query type implementing Query.
  • R: The return type, either a single ReadModel or an Iterator.

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 creates a QueryHandler from a function.

Parameters:

  • fn: The function to wrap as a QueryHandler.

Returns:

  • QueryHandler[T,R]: A handler that implements QueryHandler interface.

Example Usage: 

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

type Revision 

type Revision uint64

Revision matches exactly a numeric revision.

func (Revision) ToRawInt64 

func (r Revision) ToRawInt64() int64

type StreamExists 

type StreamExists struct{}

StreamExists means the stream must exist.

func (StreamExists) ToRawInt64 

func (StreamExists) ToRawInt64() int64

type StreamNamer 

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

StreamNamer produces the stream name for a given command, with access to context 

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

DefaultStreamNamer is the default function used to determine the stream name for a given command when no custom StreamNamer is provided.

By default, it returns the AggregateID of the command as the stream name.

This variable can be overridden globally to change the default behavior for all command handlers, for example to support multi-tenancy, prefixes, or other custom naming conventions.

Example usage: 

// Default behavior uses AggregateID
stream := DefaultStreamNamer(ctx, myCommand)

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

type StreamRevisionConflictError 

type StreamRevisionConflictError struct {
	Stream           string
	ExpectedRevision StreamState
	ActualRevision   StreamState
}

func (StreamRevisionConflictError) Error 

func (s StreamRevisionConflictError) Error() string

type StreamState 

type StreamState interface {
	ToRawInt64() int64
}

type SubscriberOption 

type SubscriberOption func(cfg any)

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.