pgeventstore

README

pgeventstore

Event store using Postgres as the storage engine, ported (back) from oraeventstore.

Database Set Up

For the database set up, we assume the availability of a database user who can create schema objects.

To install the schema, use flyway to install the schema. Installation involves downloading the schema and dropping the postgres JDBC jar into the flyway drivers directory.

Edit the flyway.conf in the db directory with your particulars, then from the db directory run:

flyway -user=esdbo -password=password -locations=filesystem:migration migrate

A Note on the Publish Table

The publish table simply writes the aggregate IDs of recently stored aggregates, which picks up creation and updates. Another process will need to read from the table to pick up the published aggregate, read the actual data from the event store table, do something with it (publish it to a queue, write out CQRS query views, etc), then delete the record from the publish table.

Consider the use of the pgpublish project which publishes records written to the publich table to an AWS SNS topic.

Dependencies

go get github.com/xtracdev/goes
go get github.com/Sirupsen/logrus
go get github.com/gucumber/gucumber
go get github.com/stretchr/testify/assert
go get github.com/lib/pq

Contributing

To contribute, you must certify you agree with the Developer Certificate of Origin by signing your commits via git -s. To create a signature, configure your user name and email address in git. Sign with your real name, do not use pseudonyms or submit anonymous commits.

In terms of workflow:

1. For significant changes or improvement, create an issue before commencing work.
2. Fork the respository, and create a branch for your edits.
3. Add tests that cover your changes, unit tests for smaller changes, acceptance test for more significant functionality.
4. Run gofmt on each file you change before committing your changes.
5. Run golint on each file you change before committing your changes.
6. Make sure all the tests pass before committing your changes.
7. Commit your changes and issue a pull request.

License

(c) 2017 Fidelity Investments Licensed under the Apache License, Version 2.0

Documentation

Index

• Variables
• type PGEventStore
  • func NewPGEventStore(db *sql.DB, enableEventPublishing bool) (*PGEventStore, error)
  • func (es *PGEventStore) GetMaxVersionForAggregate(aggID string) (*int, error)
  • func (es *PGEventStore) RepublishAllEvents() error
  • func (es *PGEventStore) RetrieveEvents(aggID string) ([]goes.Event, error)
  • func (es *PGEventStore) StoreEvents(agg *goes.Aggregate) error

Variables

var (
	// ErrConcurrency is produced when version conflicts arise from uncoordinated write occur
	ErrConcurrency = errors.New("Concurrency Exception")
	// ErrPayloadType is produced when an unsupported payload type is used
	ErrPayloadType = errors.New("Only payloads of type []byte are allowed")
	// ErrEventInsert is produced when there's an error inserting into the events table
	ErrEventInsert = errors.New("Error inserting record into events table")
	// ErrPubInsert is oridyced when there's an error inserting into the publish table
	ErrPubInsert = errors.New("Error inserting record into pub table")
)

Types

type PGEventStore

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

PGEventStore is the type encapsulating event store functionality implemented using Postgres as the durable event store.

func NewPGEventStore

func NewPGEventStore(db *sql.DB, enableEventPublishing bool) (*PGEventStore, error)

NewPGEventStore instantiates an event store with the provided database interface instance. Passing true via enableEventPublishing writes events to the publish table when transactions commit.

func (*PGEventStore) GetMaxVersionForAggregate

func (es *PGEventStore) GetMaxVersionForAggregate(aggID string) (*int, error)

GetMaxVersionForAggregate returns the greatest version number for the given aggregate.

func (*PGEventStore) RepublishAllEvents

func (es *PGEventStore) RepublishAllEvents() error

RepublishAllEvents writes all the events in the event store to the publish table.

func (*PGEventStore) RetrieveEvents

func (es *PGEventStore) RetrieveEvents(aggID string) ([]goes.Event, error)

RetrieveEvents returns the events associated with the given aggregate.

func (*PGEventStore) StoreEvents

func (es *PGEventStore) StoreEvents(agg *goes.Aggregate) error

StoreEvents stores the uncommitted events assocaited with the given aggregate.