event_store_adapter_go

package module
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Sep 7, 2023 License: MIT Imports: 10 Imported by: 0

README

event-store-adapter-go

CI Release tokei

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.StoreEventAndSnapshotOpt(event, version, aggregate)
}

func (r *UserAccountRepository) FindById(id AggregateId) (*UserAccount, error) {
    result, err := r.eventStore.GetLatestSnapshotById(id, r.aggregateConverter)
    if err != nil {
        return nil, err
    }
    events, err := r.eventStore.GetEventsByIdSinceSeqNr(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

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
}

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) GetEventsByIdSinceSeqNr added in v0.3.0 

func (es *EventStore) GetEventsByIdSinceSeqNr(aggregateId AggregateId, seqNr uint64, converter EventConverter) ([]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 (*EventStore) GetLatestSnapshotById added in v0.3.0 

func (es *EventStore) GetLatestSnapshotById(aggregateId AggregateId, converter AggregateConverter) (*AggregateWithSeqNrWithVersion, 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 (*EventStore) StoreEventAndSnapshotOpt added in v0.3.0 

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

StoreEventAndSnapshotOpt 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 WithDeleteTtl added in v0.3.0 

func WithDeleteTtl(deleteTtl time.Duration) EventStoreOption

func WithEventSerializer 

func WithEventSerializer(eventSerializer EventSerializer) EventStoreOption

WithEventSerializer sets an event serializer.

func WithKeepSnapshot added in v0.3.0 

func WithKeepSnapshot(keepSnapshot bool) EventStoreOption

func WithKeepSnapshotCount added in v0.3.0 

func WithKeepSnapshotCount(keepSnapshotCount uint32) EventStoreOption

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 PkeySkey added in v0.3.0 

type PkeySkey struct {
	Pkey string
	Skey string
}

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.