fix: properly handle otp sms challenge notification in session api (#7653)

* fix: properly handle otp sms challenge notification in session api

* small fix

internal/notification/handlers/already_handled.go

@@ -6,21 +6,45 @@ import (
 	"github.com/zitadel/zitadel/internal/eventstore"
 )
 
-func (n *NotificationQueries) IsAlreadyHandled(ctx context.Context, event eventstore.Event, data map[string]interface{}, aggregateType eventstore.AggregateType, eventTypes ...eventstore.EventType) (bool, error) {
-	events, err := n.es.Filter(
-		ctx,
-		eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
-			InstanceID(event.Aggregate().InstanceID).
-			SequenceGreater(event.Sequence()).
-			AddQuery().
-			AggregateTypes(aggregateType).
-			AggregateIDs(event.Aggregate().ID).
-			EventTypes(eventTypes...).
-			EventData(data).
-			Builder(),
-	)
+type alreadyHandled struct {
+	event      eventstore.Event
+	eventTypes []eventstore.EventType
+	data       map[string]interface{}
+
+	handled bool
+}
+
+func (a *alreadyHandled) Reduce() error {
+	return nil
+}
+
+func (a *alreadyHandled) AppendEvents(event ...eventstore.Event) {
+	if len(event) > 0 {
+		a.handled = true
+	}
+}
+
+func (a *alreadyHandled) Query() *eventstore.SearchQueryBuilder {
+	return eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
+		InstanceID(a.event.Aggregate().InstanceID).
+		SequenceGreater(a.event.Sequence()).
+		AddQuery().
+		AggregateTypes(a.event.Aggregate().Type).
+		AggregateIDs(a.event.Aggregate().ID).
+		EventTypes(a.eventTypes...).
+		EventData(a.data).
+		Builder()
+}
+
+func (n *NotificationQueries) IsAlreadyHandled(ctx context.Context, event eventstore.Event, data map[string]interface{}, eventTypes ...eventstore.EventType) (bool, error) {
+	already := &alreadyHandled{
+		event:      event,
+		eventTypes: eventTypes,
+		data:       data,
+	}
+	err := n.es.FilterToQueryReducer(ctx, already)
 	if err != nil {
 		return false, err
 	}
-	return len(events) > 0, nil
+	return already.handled, nil
 }