event_store_adapter_go

package module
v1.0.2 Latest Latest
Warning

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

 Go to latest Published: Jan 20, 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.

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() AggregateId
	GetSeqNr() uint64
	GetVersion() uint64
	WithVersion(version uint64) Aggregate
}

type AggregateConverter 

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

type AggregateId 

type AggregateId interface {
	fmt.Stringer
	GetTypeName() string
	GetValue() string
	AsString() string
}

type AggregateResult added in v0.3.4 

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

func (*AggregateResult) Aggregate added in v0.3.4 

func (awsv *AggregateResult) Aggregate() Aggregate

func (*AggregateResult) Empty added in v0.3.4 

func (awsv *AggregateResult) Empty() bool

func (*AggregateResult) Present added in v0.3.4 

func (awsv *AggregateResult) Present() bool

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
}

func NewDeserializationError added in v0.4.0 

func NewDeserializationError(message string, cause error) *DeserializationError

type Event 

type Event interface {
	fmt.Stringer
	GetId() string
	GetTypeName() string
	GetAggregateId() AggregateId
	GetSeqNr() uint64
	IsCreated() bool
	GetOccurredAt() uint64
}

type EventConverter 

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

type EventSerializer 

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

type EventStore 

type EventStore interface {
	GetLatestSnapshotById(aggregateId AggregateId) (*AggregateResult, error)
	GetEventsByIdSinceSeqNr(aggregateId AggregateId, seqNr uint64) ([]Event, error)
	PersistEvent(event Event, version uint64) error
	PersistEventAndSnapshot(event Event, aggregate Aggregate) error
}

type EventStoreBaseError added in v0.3.7 

type EventStoreBaseError struct {
	Message string
	Cause   error
}

func (*EventStoreBaseError) Error added in v0.3.7 

func (e *EventStoreBaseError) Error() string

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.

func (*EventStoreOnDynamoDB) GetEventsByIdSinceSeqNr added in v1.0.0 

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

GetEventsByIdSinceSeqNr returns events by aggregateId and seqNr.

- aggregateId is an aggregateId to get events. - seqNr is a seqNr to get events. - converter is a converter to convert a map to an event.

Returns events and an error.

func (*EventStoreOnDynamoDB) GetLatestSnapshotById added in v1.0.0 

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

GetLatestSnapshotById returns a snapshot by aggregateId.

- aggregateId is an aggregateId to get a snapshot. - converter is a converter to convert a map to an aggregate.

Returns a snapshot and an 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
}

func NewEventStoreOnMemory added in v1.0.0 

func NewEventStoreOnMemory() *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 a delete ttl.

- 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.

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.

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.

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.

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.

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.

Returns an EventStoreOption.

type IOError added in v0.3.7 

type IOError struct {
	EventStoreBaseError
}

func NewIOError added in v0.4.0 

func NewIOError(message string, cause error) *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
}

func NewOptimisticLockError added in v0.4.0 

func NewOptimisticLockError(message string, cause error) *OptimisticLockError

type SerializationError added in v0.3.7 

type SerializationError struct {
	EventStoreBaseError
}

func NewSerializationError added in v0.4.0 

func NewSerializationError(message string, cause error) *SerializationError

type SnapshotSerializer 

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

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.