ignore client ids

internal/eventstore/handler/crdb/handler_stmt.go

@@ -128,16 +128,18 @@ func (h *StatementHandler) SearchQuery(ctx context.Context, instanceIDs []string
 					break
 				}
 			}
+			if h.reduceScheduledPseudoEvent {
+				queryBuilder.
+					AddQuery().
+					SequenceGreater(seq).
+					InstanceID(instanceID)
+				continue
+			}
 			queryBuilder.
 				AddQuery().
 				SequenceGreater(seq).
-				InstanceID(instanceID)
-
-			if !h.reduceScheduledPseudoEvent {
-				queryBuilder.
-					AddQuery().
-					AggregateTypes(aggregateType)
-			}
+				InstanceID(instanceID).
+				AggregateTypes(aggregateType)
 		}
 	}
 

internal/eventstore/handler/crdb/statement.go

@@ -315,6 +315,16 @@ func NewLessThanCond(column string, value interface{}) handler.Condition {
 	}
 }
 
+func NewContainsCond(column string, value interface{}) handler.Condition {
+	return handler.Condition{
+		Name:  column,
+		Value: value,
+		ParameterOpt: func(placeholder string) string {
+			return fmt.Sprintf(" @> ARRAY[%s]", placeholder)
+		},
+	}
+}
+
 // NewCopyStatement creates a new upsert statement which updates a column from an existing row
 // cols represent the columns which are objective to change.
 // if the value of a col is empty the data will be copied from the selected row

internal/eventstore/handler/handler_projection.go

@@ -100,13 +100,13 @@ func NewProjectionHandler(
 		reduceScheduledPseudoEvent: reduceScheduledPseudoEvent,
 	}
 
-	go func(subscribe bool) {
+	go func() {
 		<-initialized
-		if subscribe {
+		if !h.reduceScheduledPseudoEvent {
 			go h.subscribe(ctx)
 		}
 		go h.schedule(ctx)
-	}(!h.reduceScheduledPseudoEvent)
+	}()
 
 	return h
 }

internal/eventstore/handler/handler_projection_test.go

@@ -342,6 +342,7 @@ func TestProjectionHandler_Process(t *testing.T) {
 				nil,
 				nil,
 				nil,
+				false,
 			)
 
 			index, err := h.Process(tt.args.ctx, tt.args.events...)