Allow to wrap Event appending function for Postgres EventStore (#59)

* Allow to wrap append to store function in postgres.EventStore

* Cleanup

* Update eventstore/postgres/store.go

Co-authored-by: Danilo Cianfrone <danilocianfr@gmail.com>

* Add documentation, fix linter issues

Co-authored-by: Konstantin Alexandrov <kale@hellofresh.com>
Co-authored-by: Danilo Cianfrone <danilocianfr@gmail.com>

Author: 3 people

CHANGELOG.md

@@ -7,7 +7,7 @@ project adheres to [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 ### Added
-- ...
+- An option to override Event appending logic in Postgres EventStore implementation. 
 
 ### Changed
 - ...

eventstore/postgres/go.sum

@@ -30,8 +30,6 @@ github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymF
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5ynNVH9qI8YYLbd1fK2po=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
-github.com/get-eventually/go-eventually v0.0.0-20210830220720-a038785a5316 h1:mdvJ8lYRc0BXPnkQczyF7EeV9xNnKTMh459lnetvDus=
-github.com/get-eventually/go-eventually v0.0.0-20210830220720-a038785a5316/go.mod h1:yOBDlloZ9BP1Hbmdp0lvkCC49esxLizZuytfjakJ5J4=
 github.com/get-eventually/go-eventually v0.0.0-20210903204041-6da801fa7f4e h1:9+YQjGoKqBYE0I2GPW3HVNb6k6T6dQHkvEcLORoIOgg=
 github.com/get-eventually/go-eventually v0.0.0-20210903204041-6da801fa7f4e/go.mod h1:yOBDlloZ9BP1Hbmdp0lvkCC49esxLizZuytfjakJ5J4=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=

eventstore/postgres/store.go

@@ -20,19 +20,47 @@ var (
 	_ eventstore.SequenceNumberGetter = &EventStore{}
 )
 
+// AppendToStoreFunc represents a function type for persisting an instance of eventually.Event in postgres.EventStore.
+type AppendToStoreFunc func(
+	ctx context.Context,
+	tx *sql.Tx,
+	id stream.ID,
+	expected eventstore.VersionCheck,
+	event eventually.Event,
+) (int64, error)
+
 // EventStore is an eventstore.Store implementation which uses
 // PostgreSQL as backend datastore.
 type EventStore struct {
-	db       *sql.DB
-	registry eventstore.Registry
+	db            *sql.DB
+	registry      eventstore.Registry
+	appendToStore AppendToStoreFunc
+}
+
+// Option defines a type for providing additional constructor adjustments for postgres.EventStore.
+type Option func(EventStore) EventStore
+
+// WithAppendMiddleware allows overriding the internal logic for appending events within a transaction.
+func WithAppendMiddleware(wrap func(AppendToStoreFunc) AppendToStoreFunc) Option {
+	return func(store EventStore) EventStore {
+		store.appendToStore = wrap(store.appendToStore)
+		return store
+	}
 }
 
 // NewEventStore creates a new EventStore using the database connection pool provided.
-func NewEventStore(db *sql.DB) EventStore {
-	return EventStore{
-		db:       db,
-		registry: eventstore.NewRegistry(json.Unmarshal),
+func NewEventStore(db *sql.DB, options ...Option) EventStore {
+	store := EventStore{
+		db:            db,
+		registry:      eventstore.NewRegistry(json.Unmarshal),
+		appendToStore: appendEvent,
 	}
+
+	for _, option := range options {
+		store = option(store)
+	}
+
+	return store
 }
 
 // Register registers Domain Events used by the application in order to decode events
@@ -129,7 +157,7 @@ func (st EventStore) Append(
 	}()
 
 	for _, event := range events {
-		if v, err = st.appendEvent(ctx, tx, id, expected, event); err != nil {
+		if v, err = st.appendToStore(ctx, tx, id, expected, event); err != nil {
 			return 0, err
 		}
 
@@ -145,7 +173,7 @@ func (st EventStore) Append(
 }
 
 // TODO(ar3s3ru): add the ErrConflict error in case of optimistic concurrency issues.
-func (st *EventStore) appendEvent(
+func appendEvent(
 	ctx context.Context,
 	tx *sql.Tx,
 	id stream.ID,

eventstore/postgres/store_test.go

@@ -40,6 +40,29 @@ func obtainEventStore(t *testing.T) (*sql.DB, postgres.EventStore) {
 	db, err := sql.Open("postgres", url)
 	require.NoError(t, err)
 
+	handleError := func(err error) {
+		if !assert.NoError(t, err) {
+			t.FailNow()
+		}
+	}
+
+	tx, err := db.Begin()
+	require.NoError(t, err)
+
+	// Reset checkpoints for subscriptions.
+	_, err = tx.Exec("DELETE FROM subscriptions_checkpoints")
+	require.NoError(t, err)
+
+	// Reset committed events and streams.
+	_, err = tx.Exec("DELETE FROM streams")
+	require.NoError(t, err)
+
+	// Reset the global sequence number to 1.
+	_, err = tx.Exec("ALTER SEQUENCE events_global_sequence_number_seq RESTART WITH 1")
+	require.NoError(t, err)
+
+	handleError(tx.Commit())
+
 	return db, postgres.NewEventStore(db)
 }
 
@@ -50,29 +73,6 @@ func TestStoreSuite(t *testing.T) {
 	require.NoError(t, store.Register(internal.IntPayload(0)))
 
 	suite.Run(t, eventstore.NewStoreSuite(func() eventstore.Store {
-		handleError := func(err error) {
-			if !assert.NoError(t, err) {
-				t.FailNow()
-			}
-		}
-
-		tx, err := db.Begin()
-		require.NoError(t, err)
-
-		// Reset checkpoints for subscriptions.
-		_, err = tx.Exec("DELETE FROM subscriptions_checkpoints")
-		require.NoError(t, err)
-
-		// Reset committed events and streams.
-		_, err = tx.Exec("DELETE FROM streams")
-		require.NoError(t, err)
-
-		// Reset the global sequence number to 1.
-		_, err = tx.Exec("ALTER SEQUENCE events_global_sequence_number_seq RESTART WITH 1")
-		require.NoError(t, err)
-
-		handleError(tx.Commit())
-
 		return store
 	}))
 }
@@ -111,3 +111,41 @@ func TestLatestSequenceNumber(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, latestSequenceNumber, actual)
 }
+
+func TestAppendToStoreWrapperOption(t *testing.T) {
+	db, _ := obtainEventStore(t)
+	defer func() { assert.NoError(t, db.Close()) }()
+
+	triggered := false
+
+	store := postgres.NewEventStore(
+		db,
+		postgres.WithAppendMiddleware(func(super postgres.AppendToStoreFunc) postgres.AppendToStoreFunc {
+			return func(
+				ctx context.Context,
+				tx *sql.Tx,
+				id stream.ID,
+				expected eventstore.VersionCheck,
+				event eventually.Event,
+			) (int64, error) {
+				triggered = true
+				return super(ctx, tx, id, expected, event)
+			}
+		}),
+	)
+
+	require.NoError(t, store.Register(internal.IntPayload(0)))
+
+	ctx := context.Background()
+
+	_, _ = store.Append(
+		ctx,
+		firstInstance,
+		eventstore.VersionCheck(int64(-1)),
+		eventually.Event{Payload: internal.IntPayload(13)},
+	)
+
+	latestSequenceNumber, _ := store.LatestSequenceNumber(ctx)
+	assert.Equal(t, int64(1), latestSequenceNumber)
+	assert.True(t, triggered)
+}