Ensure Projection Init is called

During one of the refactors we lost the Projection Init call.
This init call has been added again to ensure projections can setup there initial state.

Author: boekkooi-fresh

driver/sql/internal/notification_projector.go

@@ -18,7 +18,8 @@ type NotificationProjector struct {
 
 	storage driverSQL.ProjectionStorage
 
-	decodeProjectionState driverSQL.ProjectionStateDecoder
+	projectionStateInit   driverSQL.ProjectionStateInitializer
+	projectionStateDecode driverSQL.ProjectionStateDecoder
 	handlers              map[string]goengine.MessageHandler
 
 	eventLoader driverSQL.EventStreamLoader
@@ -31,23 +32,26 @@ type NotificationProjector struct {
 func NewNotificationProjector(
 	db *sql.DB,
 	storage driverSQL.ProjectionStorage,
-	acquireUnmarshalState driverSQL.ProjectionStateDecoder,
+	projectionStateInit driverSQL.ProjectionStateInitializer,
+	projectionStateDecode driverSQL.ProjectionStateDecoder,
 	eventHandlers map[string]goengine.MessageHandler,
 	eventLoader driverSQL.EventStreamLoader,
 	resolver goengine.MessagePayloadResolver,
 	logger goengine.Logger,
 ) (*NotificationProjector, error) {
 	switch {
 	case db == nil:
-		return nil, errors.New("db cannot be nil")
+		return nil, goengine.InvalidArgumentError("db")
 	case storage == nil:
-		return nil, errors.New("storage cannot be nil")
+		return nil, goengine.InvalidArgumentError("storage")
+	case projectionStateInit == nil:
+		return nil, goengine.InvalidArgumentError("projectionStateInit")
 	case len(eventHandlers) == 0:
-		return nil, errors.New("eventHandlers cannot be empty")
+		return nil, goengine.InvalidArgumentError("eventHandlers")
 	case eventLoader == nil:
-		return nil, errors.New("eventLoader cannot be nil")
+		return nil, goengine.InvalidArgumentError("eventLoader")
 	case resolver == nil:
-		return nil, errors.New("resolver cannot be nil")
+		return nil, goengine.InvalidArgumentError("resolver")
 	}
 
 	if logger == nil {
@@ -57,7 +61,8 @@ func NewNotificationProjector(
 	return &NotificationProjector{
 		db:                    db,
 		storage:               storage,
-		decodeProjectionState: acquireUnmarshalState,
+		projectionStateInit:   projectionStateInit,
+		projectionStateDecode: projectionStateDecode,
 		handlers:              wrapProjectionHandlers(eventHandlers),
 		eventLoader:           eventLoader,
 		resolver:              resolver,
@@ -114,10 +119,17 @@ func (s *NotificationProjector) project(
 	}
 	defer releaseLock()
 
-	// Unmarshal the projection state
+	// Decode or initialize projection state
 	var projectionState interface{}
-	if s.decodeProjectionState != nil {
-		projectionState, err = s.decodeProjectionState(rawState.ProjectionState)
+	if rawState.Position == 0 {
+		// This is the fist time the projection runs so initialize the state
+		projectionState, err = s.projectionStateInit(ctx)
+		if err != nil {
+			return err
+		}
+	} else if s.projectionStateDecode != nil {
+		// Unmarshal the projection state
+		projectionState, err = s.projectionStateDecode(rawState.ProjectionState)
 		if err != nil {
 			return err
 		}

driver/sql/postgres/projector_aggregate.go

@@ -84,6 +84,7 @@ func NewAggregateProjector(
 	executor, err := internal.NewNotificationProjector(
 		db,
 		storage,
+		projection.Init,
 		stateDecoder,
 		projection.Handlers(),
 		aggregateProjectionEventStreamLoader(eventStore, projection.FromStream(), aggregateTypeName),

driver/sql/postgres/projector_aggregate_storage.go

@@ -92,7 +92,7 @@ func newAggregateProjectionStorage(
 		queryAcquireLock: fmt.Sprintf(
 			`WITH new_projection AS (
 			  INSERT INTO %[1]s (aggregate_id, state) SELECT $1, 'null' WHERE NOT EXISTS (
-		    	SELECT * FROM %[1]s WHERE aggregate_id = $1
+		    	SELECT * FROM %[1]s WHERE aggregate_id = $1 LIMIT 1
 			  ) ON CONFLICT DO NOTHING
 			  RETURNING *
 			)
@@ -118,7 +118,7 @@ func (a *aggregateProjectionStorage) LoadOutOfSync(ctx context.Context, conn dri
 	return conn.QueryContext(ctx, a.queryOutOfSyncProjections)
 }
 
-func (a *aggregateProjectionStorage) PersistState(conn *sql.Conn, notification *driverSQL.ProjectionNotification, state driverSQL.ProjectionState) error {
+func (a *aggregateProjectionStorage) PersistState(conn driverSQL.Execer, notification *driverSQL.ProjectionNotification, state driverSQL.ProjectionState) error {
 	encodedState, err := a.projectionStateEncoder(state.ProjectionState)
 	if err != nil {
 		return err

driver/sql/postgres/projector_stream.go

@@ -3,7 +3,6 @@ package postgres
 import (
 	"context"
 	"database/sql"
-	"fmt"
 	"math"
 	"strings"
 	"sync"
@@ -17,15 +16,14 @@ import (
 // StreamProjector is a postgres projector used to execute a projection against an event stream.
 type StreamProjector struct {
 	sync.Mutex
-	executor *internalSQL.NotificationProjector
-
-	db *sql.DB
 
-	projectionName  string
-	projectionTable string
+	db       *sql.DB
+	executor *internalSQL.NotificationProjector
+	storage  *streamProjectionStorage
 
-	logger                 goengine.Logger
 	projectionErrorHandler driverSQL.ProjectionErrorCallback
+
+	logger goengine.Logger
 }
 
 // NewStreamProjector creates a new projector for a projection
@@ -75,6 +73,7 @@ func NewStreamProjector(
 	executor, err := internalSQL.NewNotificationProjector(
 		db,
 		storage,
+		projection.Init,
 		stateDecoder,
 		projection.Handlers(),
 		streamProjectionEventStreamLoader(eventStore, projection.FromStream()),
@@ -86,15 +85,11 @@ func NewStreamProjector(
 	}
 
 	return &StreamProjector{
-		executor: executor,
-
-		db: db,
-
-		projectionName:         projection.Name(),
-		projectionTable:        projectionTable,
+		db:                     db,
+		executor:               executor,
+		storage:                storage,
 		projectionErrorHandler: projectionErrorHandler,
-
-		logger: logger,
+		logger:                 logger,
 	}, nil
 }
 
@@ -110,7 +105,7 @@ func (s *StreamProjector) Run(ctx context.Context) error {
 		return nil
 	}
 
-	if err := s.setupProjection(ctx); err != nil {
+	if err := s.storage.CreateProjection(ctx, s.db); err != nil {
 		return err
 	}
 
@@ -129,7 +124,7 @@ func (s *StreamProjector) RunAndListen(ctx context.Context, listener driverSQL.L
 		return nil
 	}
 
-	if err := s.setupProjection(ctx); err != nil {
+	if err := s.storage.CreateProjection(ctx, s.db); err != nil {
 		return err
 	}
 
@@ -168,19 +163,3 @@ func (s *StreamProjector) processNotification(
 		math.MaxInt16,
 	)
 }
-
-// setupProjection Creates the projection if none exists
-func (s *StreamProjector) setupProjection(ctx context.Context) error {
-	// Ignore duplicate inserts
-	_, err := s.db.ExecContext(
-		ctx,
-		/* #nosec */
-		fmt.Sprintf(
-			`INSERT INTO %s (name) VALUES ($1) ON CONFLICT DO NOTHING`,
-			QuoteIdentifier(s.projectionTable),
-		),
-		s.projectionName,
-	)
-
-	return err
-}

driver/sql/postgres/projector_stream_storage.go

@@ -8,7 +8,6 @@ import (
 	"strings"
 
 	"github.com/hellofresh/goengine"
-
 	driverSQL "github.com/hellofresh/goengine/driver/sql"
 )
 
@@ -26,6 +25,7 @@ type streamProjectionStorage struct {
 
 	logger goengine.Logger
 
+	queryCreateProjection    string
 	queryAcquireLock         string
 	queryAcquirePositionLock string
 	queryReleaseLock         string
@@ -62,6 +62,10 @@ func newStreamProjectionStorage(
 		projectionStateEncoder: projectionStateEncoder,
 		logger:                 logger,
 
+		queryCreateProjection: fmt.Sprintf(
+			`INSERT INTO %s (name) VALUES ($1) ON CONFLICT DO NOTHING`,
+			projectionTableQuoted,
+		),
 		queryAcquireLock: fmt.Sprintf(
 			`SELECT pg_try_advisory_lock(%[2]s::regclass::oid::int, no), locked, position, state FROM %[1]s WHERE name = $1`,
 			projectionTableQuoted,
@@ -88,7 +92,12 @@ func newStreamProjectionStorage(
 	}, nil
 }
 
-func (s *streamProjectionStorage) PersistState(conn *sql.Conn, notification *driverSQL.ProjectionNotification, state driverSQL.ProjectionState) error {
+func (s *streamProjectionStorage) CreateProjection(ctx context.Context, conn driverSQL.Execer) error {
+	_, err := conn.ExecContext(ctx, s.queryCreateProjection, s.projectionName)
+	return err
+}
+
+func (s *streamProjectionStorage) PersistState(conn driverSQL.Execer, notification *driverSQL.ProjectionNotification, state driverSQL.ProjectionState) error {
 	encodedState, err := s.projectionStateEncoder(state.ProjectionState)
 	if err != nil {
 		return err

driver/sql/projection.go

@@ -29,6 +29,9 @@ type (
 		ProjectionState []byte
 	}
 
+	// ProjectionStateInitializer is a func to initialize a ProjectionState.ProjectionState
+	ProjectionStateInitializer func(ctx context.Context) (interface{}, error)
+
 	// ProjectionStateEncoder is a func to marshal the ProjectionState.ProjectionState
 	ProjectionStateEncoder func(interface{}) ([]byte, error)
 
@@ -38,7 +41,7 @@ type (
 	// ProjectionStorage is an interface for handling the projection storage
 	ProjectionStorage interface {
 		// PersistState persists the state of the projection
-		PersistState(conn *sql.Conn, notification *ProjectionNotification, state ProjectionState) error
+		PersistState(conn Execer, notification *ProjectionNotification, state ProjectionState) error
 
 		// Acquire this function is used to acquire the projection and it's projectionState
 		// A projection can only be acquired once and must be released using the returned func