event_store_adapter_go

package module
v1.0.12 Latest Latest
Warning

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

 Go to latest Published: Feb 23, 2024 License: MIT Imports: 11 Imported by: 0

README

event-store-adapter-go

CI Go project version Renovate License

This library is designed to turn DynamoDB into an Event Store for CQRS/Event Sourcing.

日本語

Usage

You can easily implement an Event Sourcing-enabled repository using EventStore.

type UserAccountRepository struct {
    eventStore         EventStore
    aggregateConverter AggregateConverter
    eventConverter     EventConverter
}

func (r *UserAccountRepository) StoreEvent(event Event, version uint64) error {
    return r.eventStore.PersistEvent(event, version)
}

func (r *UserAccountRepository) StoreEventAndSnapshot(event Event, aggregate Aggregate) error {
    return r.eventStore.PersistEventAndSnapshot(event, aggregate)
}

func (r *UserAccountRepository) findById(id esag.AggregateId) (*userAccount, error) {
	result, err := r.eventStore.GetLatestSnapshotById(id, r.aggregateConverter)
	if err != nil {
		return nil, err
	}
	if result.Empty() {
		return nil, fmt.Errorf("not found")
	} else {
		events, err := r.eventStore.GetEventsByIdSinceSeqNr(id, result.Aggregate().GetSeqNr()+1, r.eventConverter)
		if err != nil {
			return nil, err
		}
		return replayUserAccount(events, result.Aggregate().(*userAccount)), nil
	}
}

The following is an example of the repository usage

eventStore, err := NewEventStoreOnDynamoDB(dynamodbClient, "journal", "snapshot", "journal-aid-index", "snapshot-aid-index", 1)
// eventStore := NewEventStoreOnMemory() // if use repository for on-memory
if err != nil {
    return err
}
repository := NewUserAccountRepository(eventStore)

userAccount1, userAccountCreated := NewUserAccount(UserAccountId{Value: "1"}, "test")
// Store an aggregate with a create event
err = repository.StoreEvent(userAccountCreated, userAccount1.Version, userAccount1)
if err != nil {
    return err
}

// Replay the aggregate from the event store
userAccount2, err := repository.FindById(&initial.Id)
if err != nil {
    return err
}

// Execute a command on the aggregate
userAccountUpdated, userAccountRenamed := userAccount2.ChangeName("test2")

// Store the new event without a snapshot
err = repository.StoreEvent(userAccountRenamed, userAccountUpdated.Version)
// Store the new event with a snapshot
// err = repository.StoreEventAndSnapshot(userAccountRenamed, userAccountUpdated)
if err != nil {
    return err
}

Table Specifications

See docs/DATABASE_SCHEMA.md.

Migration Guide

See docs/MIGRATION_GUIDE.md.

CQRS/Event Sourcing Example

See j5ik2o/cqrs-es-example-go.

License.

MIT License. See LICENSE for details.

Other language implementations

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Aggregate 

type Aggregate interface {
	fmt.Stringer

	// GetId returns the id of the aggregate.
	GetId() AggregateId

	// GetSeqNr returns the sequence number of the aggregate.
	GetSeqNr() uint64

	// GetVersion returns the version of the aggregate.
	GetVersion() uint64

	// WithVersion returns a new aggregate with the specified version.
	WithVersion(version uint64) Aggregate
}

Aggregate is the interface that represents the aggregate of DDD.

type AggregateConverter 

type AggregateConverter func(map[string]interface{}) (Aggregate, error)

AggregateConverter is the function type that converts map[string]interface{} to Aggregate.

type AggregateId 

type AggregateId interface {
	fmt.Stringer

	// GetTypeName returns the type name of the aggregate id.
	GetTypeName() string

	// GetValue returns the value of the aggregate id.
	GetValue() string

	// AsString returns the string representation of the aggregate id.
	//
	// The string representation is {TypeName}-{Value}.
	AsString() string
}

AggregateId is the interface that represents the aggregate id of DDD.

type AggregateResult added in v0.3.4 

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

AggregateResult is the result of aggregate.

func (*AggregateResult) Aggregate added in v0.3.4 

func (a *AggregateResult) Aggregate() Aggregate

Aggregate returns the aggregate.

func (*AggregateResult) Empty added in v0.3.4 

func (a *AggregateResult) Empty() bool

Empty returns true if the aggregate is nil.

func (*AggregateResult) Present added in v0.3.4 

func (a *AggregateResult) Present() bool

Present returns true if the aggregate is not nil.

type DefaultEventSerializer added in v0.3.17 

type DefaultEventSerializer struct{}

func (*DefaultEventSerializer) Deserialize added in v0.3.17 

func (s *DefaultEventSerializer) Deserialize(data []byte, eventMap *map[string]interface{}) error

func (*DefaultEventSerializer) Serialize added in v0.3.17 

func (s *DefaultEventSerializer) Serialize(event Event) ([]byte, error)

type DefaultKeyResolver 

type DefaultKeyResolver struct {
}

func (*DefaultKeyResolver) ResolvePkey 

func (kr *DefaultKeyResolver) ResolvePkey(aggregateId AggregateId, shardCount uint64) string

func (*DefaultKeyResolver) ResolveSkey 

func (kr *DefaultKeyResolver) ResolveSkey(aggregateId AggregateId, seqNr uint64) string

type DefaultSnapshotSerializer added in v0.3.17 

type DefaultSnapshotSerializer struct{}

func (*DefaultSnapshotSerializer) Deserialize added in v0.3.17 

func (s *DefaultSnapshotSerializer) Deserialize(data []byte, aggregateMap *map[string]interface{}) error

func (*DefaultSnapshotSerializer) Serialize added in v0.3.17 

func (s *DefaultSnapshotSerializer) Serialize(aggregate Aggregate) ([]byte, error)

type DeserializationError added in v0.3.7 

type DeserializationError struct {
	EventStoreBaseError
}

DeserializationError is the error type that occurs when deserialization fails.

func NewDeserializationError added in v0.4.0 

func NewDeserializationError(message string, cause error) *DeserializationError

NewDeserializationError is the constructor of DeserializationError.

type Event 

type Event interface {
	fmt.Stringer
	// GetId returns the id of the event.
	GetId() string

	// GetTypeName returns the type name of the event.
	GetTypeName() string

	// GetAggregateId returns the aggregate id of the event.
	GetAggregateId() AggregateId

	// GetSeqNr returns the sequence number of the event.
	GetSeqNr() uint64

	// IsCreated returns true if the event is created.
	IsCreated() bool

	// GetOccurredAt returns the occurred at of the event.
	GetOccurredAt() uint64
}

Event is the interface that represents the domain event of DDD.

type EventConverter 

type EventConverter func(map[string]interface{}) (Event, error)

EventConverter is the function type that converts map[string]interface{} to Event.

type EventSerializer 

type EventSerializer interface {
	// Serialize serializes the event.
	Serialize(event Event) ([]byte, error)
	// Deserialize deserializes the event.
	Deserialize(data []byte, eventMap *map[string]interface{}) error
}

EventSerializer is an interface that serializes and deserializes events.

type EventStore 

type EventStore interface {
	// GetLatestSnapshotById returns the latest snapshot of the aggregate.
	GetLatestSnapshotById(aggregateId AggregateId) (*AggregateResult, error)
	// GetEventsByIdSinceSeqNr returns the events of the aggregate since the specified sequence number.
	GetEventsByIdSinceSeqNr(aggregateId AggregateId, seqNr uint64) ([]Event, error)
	// PersistEvent persists the event.
	PersistEvent(event Event, version uint64) error
	// PersistEventAndSnapshot persists the event and the snapshot.
	PersistEventAndSnapshot(event Event, aggregate Aggregate) error
}

EventStore is the interface for persisting events and snapshots.

type EventStoreBaseError added in v0.3.7 

type EventStoreBaseError struct {
	// Message is a message of the error.
	Message string
	// Cause is a cause of the error.
	Cause error
}

EventStoreBaseError is a base error of EventStore.

func (*EventStoreBaseError) Error added in v0.3.7 

func (e *EventStoreBaseError) Error() string

Error returns the message of the error.

type EventStoreOnDynamoDB added in v1.0.0 

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

EventStoreOnDynamoDB is EventStore for DynamoDB.

func NewEventStoreOnDynamoDB added in v1.0.0 

func NewEventStoreOnDynamoDB(
	client *dynamodb.Client,
	journalTableName string,
	snapshotTableName string,
	journalAidIndexName string,
	snapshotAidIndexName string,
	shardCount uint64,
	eventConverter EventConverter,
	snapshotConverter AggregateConverter,
	options ...EventStoreOption,
) (*EventStoreOnDynamoDB, error)

NewEventStoreOnDynamoDB returns a new EventStore.

# Parameters - client is a DynamoDB client. - journalTableName is a journal table name. - snapshotTableName is a snapshot table name. - journalAidIndexName is a journal aggregateId index name. - snapshotAidIndexName is a snapshot aggregateId index name. - shardCount is a shard count. - eventConverter is a converter to convert a map to an event. - snapshotConverter is a converter to convert a map to an aggregate. - options is an EventStoreOption.

# Returns - an EventStore - an error

func (*EventStoreOnDynamoDB) GetEventsByIdSinceSeqNr added in v1.0.0 

func (es *EventStoreOnDynamoDB) GetEventsByIdSinceSeqNr(aggregateId AggregateId, seqNr uint64) ([]Event, error)

func (*EventStoreOnDynamoDB) GetLatestSnapshotById added in v1.0.0 

func (es *EventStoreOnDynamoDB) GetLatestSnapshotById(aggregateId AggregateId) (*AggregateResult, error)

func (*EventStoreOnDynamoDB) PersistEvent added in v1.0.0 

func (es *EventStoreOnDynamoDB) PersistEvent(event Event, version uint64) error

func (*EventStoreOnDynamoDB) PersistEventAndSnapshot added in v1.0.0 

func (es *EventStoreOnDynamoDB) PersistEventAndSnapshot(event Event, aggregate Aggregate) error

type EventStoreOnMemory added in v1.0.0 

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

EventStoreOnMemory is the memory implementation of EventStore.

func NewEventStoreOnMemory added in v1.0.0 

func NewEventStoreOnMemory() *EventStoreOnMemory

NewEventStoreOnMemory is the constructor of EventStoreOnMemory.

The returned value is the pointer to EventStoreOnMemory.

func (*EventStoreOnMemory) GetEventsByIdSinceSeqNr added in v1.0.0 

func (es *EventStoreOnMemory) GetEventsByIdSinceSeqNr(aggregateId AggregateId, seqNr uint64) ([]Event, error)

func (*EventStoreOnMemory) GetLatestSnapshotById added in v1.0.0 

func (es *EventStoreOnMemory) GetLatestSnapshotById(aggregateId AggregateId) (*AggregateResult, error)

func (*EventStoreOnMemory) PersistEvent added in v1.0.0 

func (es *EventStoreOnMemory) PersistEvent(event Event, version uint64) error

func (*EventStoreOnMemory) PersistEventAndSnapshot added in v1.0.0 

func (es *EventStoreOnMemory) PersistEventAndSnapshot(event Event, aggregate Aggregate) error

type EventStoreOption 

type EventStoreOption func(*EventStoreOnDynamoDB) error

EventStoreOption is an option for EventStore.

func WithDeleteTtl added in v0.3.0 

func WithDeleteTtl(deleteTtl time.Duration) EventStoreOption

WithDeleteTtl sets the ttl for deletion snapshots.

- If you want to delete snapshots, specify a time.Duration. - If you do not want to delete snapshots, specify math.MaxInt64. - The default is math.MaxInt64.

# Parameters - deleteTtl is the time to live for deletion snapshots.

# Returns - an EventStoreOption.

func WithEventSerializer 

func WithEventSerializer(eventSerializer EventSerializer) EventStoreOption

WithEventSerializer sets an event serializer.

- If you want to change the event serializer, specify an EventSerializer. - The default is DefaultEventSerializer.

# Parameters - eventSerializer is an event serializer.

# Returns - an EventStoreOption.

func WithKeepSnapshot added in v0.3.0 

func WithKeepSnapshot(keepSnapshot bool) EventStoreOption

WithKeepSnapshot sets whether or not to keep snapshots.

- If you want to keep snapshots, specify true. - If you do not want to keep snapshots, specify false. - The default is false.

# Parameters - keepSnapshot is whether or not to keep snapshots.

# Returns - an EventStoreOption.

func WithKeepSnapshotCount added in v0.3.0 

func WithKeepSnapshotCount(keepSnapshotCount uint32) EventStoreOption

WithKeepSnapshotCount sets a keep snapshot count.

- If you want to keep snapshots, specify a keep snapshot count. - If you do not want to keep snapshots, specify math.MaxInt64. - The default is math.MaxInt64.

# Parameters - keepSnapshotCount is a keep snapshot count.

# Returns - an EventStoreOption.

func WithKeyResolver 

func WithKeyResolver(keyResolver KeyResolver) EventStoreOption

WithKeyResolver sets a key resolver.

- If you want to change the key resolver, specify a KeyResolver. - The default is DefaultKeyResolver.

# Parameters - keyResolver is a key resolver.

# Returns - an EventStoreOption.

func WithSnapshotSerializer 

func WithSnapshotSerializer(snapshotSerializer SnapshotSerializer) EventStoreOption

WithSnapshotSerializer sets a snapshot serializer.

- If you want to change the snapshot serializer, specify a SnapshotSerializer. - The default is DefaultSnapshotSerializer.

# Parameters - snapshotSerializer is a snapshot serializer.

# Returns - an EventStoreOption.

type IOError added in v0.3.7 

type IOError struct {
	EventStoreBaseError
}

IOError is the error type that occurs when IO fails.

func NewIOError added in v0.4.0 

func NewIOError(message string, cause error) *IOError

NewIOError is the constructor of IOError.

type KeyResolver 

type KeyResolver interface {
	ResolvePkey(aggregateId AggregateId, shardCount uint64) string
	ResolveSkey(aggregateId AggregateId, seqNr uint64) string
}

type OptimisticLockError added in v0.3.17 

type OptimisticLockError struct {
	EventStoreBaseError
}

OptimisticLockError is an error that occurs when the version of the aggregate does not match.

func NewOptimisticLockError added in v0.4.0 

func NewOptimisticLockError(message string, cause error) *OptimisticLockError

NewOptimisticLockError is the constructor of OptimisticLockError.

type SerializationError added in v0.3.7 

type SerializationError struct {
	EventStoreBaseError
}

SerializationError is the error type that occurs when serialization fails.

func NewSerializationError added in v0.4.0 

func NewSerializationError(message string, cause error) *SerializationError

NewSerializationError is the constructor of SerializationError.

type SnapshotSerializer 

type SnapshotSerializer interface {
	// Serialize serializes the aggregate.
	Serialize(aggregate Aggregate) ([]byte, error)
	// Deserialize deserializes the aggregate.
	Deserialize(data []byte, aggregateMap *map[string]interface{}) error
}

SnapshotSerializer is an interface that serializes and deserializes snapshots.

Source Files

View all Source files

Directories

Path Synopsis

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.