event_store_adapter_go

README

event-store-adapter-go

This library is designed to turn DynamoDB into an Event Store for 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) Store(event Event, version uint64, aggregate Aggregate) error {
    return r.eventStore.StoreEventWithSnapshot(event, version, aggregate)
}

func (r *UserAccountRepository) FindById(id AggregateId) (*UserAccount, error) {
    result, err := r.eventStore.GetSnapshotById(id, r.aggregateConverter)
    if err != nil {
        return nil, err
    }
    events, err := r.eventStore.GetEventsByIdAndSeqNr(id, result.SeqNr, r.eventConverter)
    if err != nil {
        return nil, err
    }
    return ReplayUserAccount(events, result.Aggregate.(*UserAccount), result.Version), nil
}

The following is an example of the repository usage

eventStore := NewEventStore(dynamodbClient, "journal", "snapshot", "journal-aid-index", "snapshot-aid-index", 1)
repository := NewUserAccountRepository(eventStore)

userAccount1, userAccountCreated := NewUserAccount(UserAccountId{Value: "1"}, "test")
// Store an aggregate with a create event
err = repository.Store(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, userAccountReanmed := userAccount2.ChangeName("test2")

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

Table Specifications

See docs/DATABASE_SCHEMA.md.

License.

MIT License. See LICENSE for details.

Documentation

Index

• type Aggregate
• type AggregateConverter
• type AggregateId
• type AggregateWithSeqNrWithVersion
• type DefaultKeyResolver
  • func (kr *DefaultKeyResolver) ResolvePkey(aggregateId AggregateId, shardCount uint64) string
  • func (kr *DefaultKeyResolver) ResolveSkey(aggregateId AggregateId, seqNr uint64) string
• type Event
• type EventConverter
• type EventSerializer
• type EventStore
  • func NewEventStore(client *dynamodb.Client, journalTableName string, snapshotTableName string, ...) (*EventStore, error)
  • func (es *EventStore) GetEventsByIdAndSeqNr(aggregateId AggregateId, seqNr uint64, converter EventConverter) ([]Event, error)
  • func (es *EventStore) GetSnapshotById(aggregateId AggregateId, converter AggregateConverter) (*AggregateWithSeqNrWithVersion, error)
  • func (es *EventStore) StoreEventWithSnapshot(event Event, version uint64, aggregate Aggregate) error
• type EventStoreOption
  • func WithEventSerializer(eventSerializer EventSerializer) EventStoreOption
  • func WithKeyResolver(keyResolver KeyResolver) EventStoreOption
  • func WithSnapshotSerializer(snapshotSerializer SnapshotSerializer) EventStoreOption
• type JsonEventSerializer
  • func (s *JsonEventSerializer) Deserialize(data []byte, eventMap *map[string]interface{}) error
  • func (s *JsonEventSerializer) Serialize(event Event) ([]byte, error)
• type JsonSnapshotSerializer
  • func (s *JsonSnapshotSerializer) Deserialize(data []byte, aggregateMap *map[string]interface{}) error
  • func (s *JsonSnapshotSerializer) Serialize(aggregate Aggregate) ([]byte, error)
• type KeyResolver
• type SnapshotSerializer

Types

type Aggregate

type Aggregate interface {
	fmt.Stringer
	GetId() AggregateId
}

type AggregateConverter

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

type AggregateId

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

type AggregateWithSeqNrWithVersion

type AggregateWithSeqNrWithVersion struct {
	Aggregate Aggregate
	SeqNr     uint64
	Version   uint64
}

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 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 struct {
	// contains filtered or unexported fields
}

EventStore for DynamoDB.

func NewEventStore

func NewEventStore(
	client *dynamodb.Client,
	journalTableName string,
	snapshotTableName string,
	journalAidIndexName string,
	snapshotAidIndexName string,
	shardCount uint64,
	options ...EventStoreOption,
) (*EventStore, error)

NewEventStore returns a new EventStore.

func (*EventStore) GetEventsByIdAndSeqNr

func (es *EventStore) GetEventsByIdAndSeqNr(aggregateId AggregateId, seqNr uint64, converter EventConverter) ([]Event, error)

GetEventsByIdAndSeqNr 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 (*EventStore) GetSnapshotById

func (es *EventStore) GetSnapshotById(aggregateId AggregateId, converter AggregateConverter) (*AggregateWithSeqNrWithVersion, error)

GetSnapshotById 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 (*EventStore) StoreEventWithSnapshot

func (es *EventStore) StoreEventWithSnapshot(event Event, version uint64, aggregate Aggregate) error

StoreEventWithSnapshot stores an event and a snapshot atomically.

- event is an event to store. - version is a version of the aggregate. - aggregate is an aggregate to store.

• Required when event is created, otherwise you can choose whether or not to save a snapshot.
• If you do not want to save snapshots, specify nil.

Occurs an error, if the event and the snapshot can not stored.

type EventStoreOption

type EventStoreOption func(*EventStore) error

EventStoreOption is an option for EventStore.

func WithEventSerializer

func WithEventSerializer(eventSerializer EventSerializer) EventStoreOption

WithEventSerializer sets an event serializer.

func WithKeyResolver

func WithKeyResolver(keyResolver KeyResolver) EventStoreOption

WithKeyResolver sets a key resolver.

func WithSnapshotSerializer

func WithSnapshotSerializer(snapshotSerializer SnapshotSerializer) EventStoreOption

WithSnapshotSerializer sets a snapshot serializer.

type JsonEventSerializer

type JsonEventSerializer struct{}

func (*JsonEventSerializer) Deserialize

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

func (*JsonEventSerializer) Serialize

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

type JsonSnapshotSerializer

type JsonSnapshotSerializer struct{}

func (*JsonSnapshotSerializer) Deserialize

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

func (*JsonSnapshotSerializer) Serialize

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

type KeyResolver

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

type SnapshotSerializer

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