fix text array contains

e2e/config/host.docker.internal/zitadel.yaml

@@ -9,19 +9,27 @@ Telemetry:
 ExternalDomain: host.docker.internal
 ExternalSecure: false
 
+TLS:
+  Enabled: false
+
 Database:
   cockroach:
     # This makes the e2e config reusable with an out-of-docker zitadel process and an /etc/hosts entry
     Host:  host.docker.internal
 
-TLS:
-  Enabled: false
-
 FirstInstance:
   Org:
     Human:
       PasswordChangeRequired: false
 
+DefaultInstance:
+  Org:
+    Human:
+      PasswordChangeRequired: false
+      Password: "Password1!"
+  LoginPolicy:
+    MfaInitSkipLifetime: "0"
+
 LogStore:
   Access:
     Database:
@@ -50,10 +58,6 @@ Projections:
     Telemetry:
       RequeueEvery: 1s
 
-DefaultInstance:
-  LoginPolicy:
-    MfaInitSkipLifetime: "0"
-
 SystemAPIUsers:
 - cypress:
     KeyData: "LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUlJQklqQU5CZ2txaGtpRzl3MEJBUUVGQUFPQ0FROEFNSUlCQ2dLQ0FRRUF6aStGRlNKTDdmNXl3NEtUd3pnTQpQMzRlUEd5Y20vTStrVDBNN1Y0Q2d4NVYzRWFESXZUUUtUTGZCYUVCNDV6YjlMdGpJWHpEdzByWFJvUzJoTzZ0CmgrQ1lRQ3ozS0N2aDA5QzBJenhaaUIySVMzSC9hVCs1Qng5RUZZK3ZuQWtaamNjYnlHNVlOUnZtdE9sbnZJZUkKSDdxWjB0RXdrUGZGNUdFWk5QSlB0bXkzVUdWN2lvZmRWUVMxeFJqNzMrYU13NXJ2SDREOElkeWlBQzNWZWtJYgpwdDBWajBTVVgzRHdLdG9nMzM3QnpUaVBrM2FYUkYwc2JGaFFvcWRKUkk4TnFnWmpDd2pxOXlmSTV0eXhZc3duCitKR3pIR2RIdlczaWRPRGxtd0V0NUsycGFzaVJJV0syT0dmcSt3MEVjbHRRSGFidXFFUGdabG1oQ2tSZE5maXgKQndJREFRQUIKLS0tLS1FTkQgUFVCTElDIEtFWS0tLS0tCg=="

e2e/config/localhost/zitadel.yaml

@@ -10,19 +10,27 @@ Telemetry:
 ExternalDomain: localhost
 ExternalSecure: false
 
+TLS:
+  Enabled: false
+
 Database:
   cockroach:
     # This makes the e2e config reusable with an out-of-docker zitadel process and an /etc/hosts entry
     Host: host.docker.internal
 
-TLS:
-  Enabled: false
-
 FirstInstance:
   Org:
     Human:
       PasswordChangeRequired: false
 
+DefaultInstance:
+  Org:
+    Human:
+      PasswordChangeRequired: false
+      Password: "Password1!"
+  LoginPolicy:
+    MfaInitSkipLifetime: "0"
+
 LogStore:
   Access:
     Database:
@@ -51,10 +59,6 @@ Projections:
     Telemetry:
       RequeueEvery: 30s
 
-DefaultInstance:
-  LoginPolicy:
-    MfaInitSkipLifetime: "0"
-
 SystemAPIUsers:
 - cypress:
     KeyData: "LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUlJQklqQU5CZ2txaGtpRzl3MEJBUUVGQUFPQ0FROEFNSUlCQ2dLQ0FRRUF6aStGRlNKTDdmNXl3NEtUd3pnTQpQMzRlUEd5Y20vTStrVDBNN1Y0Q2d4NVYzRWFESXZUUUtUTGZCYUVCNDV6YjlMdGpJWHpEdzByWFJvUzJoTzZ0CmgrQ1lRQ3ozS0N2aDA5QzBJenhaaUIySVMzSC9hVCs1Qng5RUZZK3ZuQWtaamNjYnlHNVlOUnZtdE9sbnZJZUkKSDdxWjB0RXdrUGZGNUdFWk5QSlB0bXkzVUdWN2lvZmRWUVMxeFJqNzMrYU13NXJ2SDREOElkeWlBQzNWZWtJYgpwdDBWajBTVVgzRHdLdG9nMzM3QnpUaVBrM2FYUkYwc2JGaFFvcWRKUkk4TnFnWmpDd2pxOXlmSTV0eXhZc3duCitKR3pIR2RIdlczaWRPRGxtd0V0NUsycGFzaVJJV0syT0dmcSt3MEVjbHRRSGFidXFFUGdabG1oQ2tSZE5maXgKQndJREFRQUIKLS0tLS1FTkQgUFVCTElDIEtFWS0tLS0tCg=="

internal/command/milestone.go

@@ -17,6 +17,6 @@ func (c *Commands) MilestonePushed(
 	if err != nil {
 		return err
 	}
-	_, err = c.eventstore.Push(ctx, milestone.NewPushedEvent(ctx, milestone.NewAggregate(ctx, id), msType, endpoints, primaryDomain))
+	_, err = c.eventstore.Push(ctx, milestone.NewPushedEvent(ctx, milestone.NewAggregate(ctx, id), msType, endpoints, primaryDomain, c.externalDomain))
 	return err
 }

internal/eventstore/handler/crdb/handler_stmt.go

@@ -137,9 +137,9 @@ func (h *StatementHandler) SearchQuery(ctx context.Context, instanceIDs []string
 			}
 			queryBuilder.
 				AddQuery().
+				AggregateTypes(aggregateType).
 				SequenceGreater(seq).
-				InstanceID(instanceID).
+				InstanceID(instanceID)
-				AggregateTypes(aggregateType)
 		}
 	}
 

internal/eventstore/handler/crdb/statement.go

@@ -236,12 +236,6 @@ func AddDeleteStatement(conditions []handler.Condition, opts ...execOption) func
 	}
 }
 
-func AddCopyStatement(conflict, from, to []handler.Column, conditions []handler.Condition, opts ...execOption) func(eventstore.Event) Exec {
-	return func(event eventstore.Event) Exec {
-		return NewCopyStatement(event, conflict, from, to, conditions, opts...).Execute
-	}
-}
-
 func NewArrayAppendCol(column string, value interface{}) handler.Column {
 	return handler.Column{
 		Name:  column,
@@ -286,42 +280,31 @@ func NewCopyCol(column, from string) handler.Column {
 	}
 }
 
-func NewIsNullCond(column string) handler.Condition {
-	return handler.Condition{
-		Name: column,
-		ParameterOpt: func(string) string {
-			return " IS NULL"
-		},
-	}
-}
-
-func NewNotEqualCond(column, value string) handler.Condition {
-	return handler.Condition{
-		Name:  column,
-		Value: value,
-		ParameterOpt: func(param string) string {
-			return fmt.Sprintf(" != %s", param)
-		},
-	}
-}
-
 func NewLessThanCond(column string, value interface{}) handler.Condition {
-	return handler.Condition{
+	return func(param string) (string, interface{}) {
-		Name:  column,
+		return column + " < " + param, value
-		Value: value,
-		ParameterOpt: func(placeholder string) string {
-			return " < " + placeholder
-		},
 	}
 }
 
-func NewContainsCond(column string, value interface{}) handler.Condition {
+func NewIsNullCond(column string) handler.Condition {
-	return handler.Condition{
+	return func(param string) (string, interface{}) {
-		Name:  column,
+		return column + " IS NULL", nil
-		Value: value,
+	}
-		ParameterOpt: func(placeholder string) string {
+}
-			return fmt.Sprintf(" @> ARRAY[%s]", placeholder)
+
-		},
+// NewTextArrayContainsCond returns a handler.Condition that checks if the column that stores an array of text contains the given value
+func NewTextArrayContainsCond(column string, value string) handler.Condition {
+	return func(param string) (string, interface{}) {
+		return column + " @> " + param, database.StringArray{value}
+	}
+}
+
+// Not is a function instead of a method, so that calling it is well readable
+// For example conditions := []handler.Condition{ Not(NewTextArrayContainsCond())}
+func Not(condition handler.Condition) handler.Condition {
+	return func(param string) (string, interface{}) {
+		cond, value := condition(param)
+		return "NOT ( " + cond + " )", value
 	}
 }
 
@@ -330,7 +313,7 @@ func NewContainsCond(column string, value interface{}) handler.Condition {
 // if the value of a col is empty the data will be copied from the selected row
 // if the value of a col is not empty the data will be set by the static value
 // conds represent the conditions for the selection subquery
-func NewCopyStatement(event eventstore.Event, conflictCols, from, to []handler.Column, conds []handler.Condition, opts ...execOption) *handler.Statement {
+func NewCopyStatement(event eventstore.Event, conflictCols, from, to, whereEqual []handler.Column, opts ...execOption) *handler.Statement {
 	columnNames := make([]string, len(to))
 	selectColumns := make([]string, len(from))
 	updateColumns := make([]string, len(columnNames))
@@ -350,8 +333,8 @@ func NewCopyStatement(event eventstore.Event, conflictCols, from, to []handler.C
 
 	}
 
-	wheres := make([]string, len(conds))
+	wheres := make([]string, len(whereEqual))
-	for i, cond := range conds {
+	for i, cond := range whereEqual {
 		argCounter++
 		wheres[i] = "copy_table." + cond.Name + " = $" + strconv.Itoa(argCounter)
 		args = append(args, cond.Value)
@@ -370,7 +353,7 @@ func NewCopyStatement(event eventstore.Event, conflictCols, from, to []handler.C
 		config.err = handler.ErrNoValues
 	}
 
-	if len(conds) == 0 {
+	if len(whereEqual) == 0 {
 		config.err = handler.ErrNoCondition
 	}
 
@@ -424,21 +407,16 @@ func columnsToQuery(cols []handler.Column) (names []string, parameters []string,
 	return names, parameters, values[:parameterIndex]
 }
 
-func conditionsToWhere(cols []handler.Condition, paramOffset int) (wheres []string, values []interface{}) {
+func conditionsToWhere(conditions []handler.Condition, paramOffset int) (wheres []string, values []interface{}) {
-	wheres = make([]string, len(cols))
+	wheres = make([]string, len(conditions))
-	values = make([]interface{}, 0, len(cols))
+	values = make([]interface{}, 0, len(conditions))
-
+	for i, conditionFunc := range conditions {
-	for i, col := range cols {
+		condition, value := conditionFunc("$" + strconv.Itoa(i+1+paramOffset))
-		param := "$" + strconv.Itoa(i+1+paramOffset)
+		wheres[i] = fmt.Sprintf("(%s)", condition)
-		wheres[i] = "(" + col.Name + " = " + param + ")"
+		if value != nil {
-		if col.ParameterOpt != nil {
+			values = append(values, value)
-			wheres[i] = "(" + col.Name + col.ParameterOpt(param) + ")"
-		}
-		if col.Value != nil {
-			values = append(values, col.Value)
 		}
 	}
-
 	return wheres, values
 }
 

internal/eventstore/handler/handler_projection.go

@@ -118,10 +118,6 @@ func (h *ProjectionHandler) Trigger(ctx context.Context, instances ...string) er
 	if len(instances) > 0 {
 		ids = instances
 	}
-	return h.processEvents(ctx, ids...)
-}
-
-func (h *ProjectionHandler) processEvents(ctx context.Context, ids ...string) error {
 	for {
 		events, hasLimitExceeded, err := h.FetchEvents(ctx, ids...)
 		if err != nil {

internal/eventstore/handler/statement.go

@@ -4,6 +4,7 @@ import (
 	"database/sql"
 	"encoding/json"
 	"errors"
+	"fmt"
 
 	"github.com/zitadel/logging"
 
@@ -62,11 +63,10 @@ func NewJSONCol(name string, value interface{}) Column {
 	return NewCol(name, marshalled)
 }
 
-type Condition Column
+type Condition func(param string) (string, interface{})
 
 func NewCond(name string, value interface{}) Condition {
-	return Condition{
+	return func(param string) (string, interface{}) {
-		Name:  name,
+		return fmt.Sprintf("%s = %s", name, param), value
-		Value: value,
 	}
 }

internal/notification/handlers/telemetry_pusher.go

@@ -20,6 +20,7 @@ import (
 	"github.com/zitadel/zitadel/internal/notification/types"
 	"github.com/zitadel/zitadel/internal/query"
 	"github.com/zitadel/zitadel/internal/query/projection"
+	"github.com/zitadel/zitadel/internal/repository/milestone"
 	"github.com/zitadel/zitadel/internal/repository/pseudo"
 )
 
@@ -132,6 +133,13 @@ func (t *telemetryPusher) pushMilestones(event eventstore.Event) (*handler.State
 
 func (t *telemetryPusher) pushMilestone(ctx context.Context, event *pseudo.ScheduledEvent, ms *query.Milestone) error {
 	ctx = authz.WithInstanceID(ctx, ms.InstanceID)
+	alreadyHandled, err := t.queries.IsAlreadyHandled(ctx, event, map[string]interface{}{"type": ms.Type}, milestone.AggregateType, milestone.PushedEventType)
+	if err != nil {
+		return err
+	}
+	if alreadyHandled {
+		return nil
+	}
 	for _, endpoint := range t.endpoints {
 		if err := types.SendJSON(
 			ctx,

internal/query/projection/label_policy.go

@@ -402,10 +402,10 @@ func (p *labelPolicyProjection) reduceActivated(event eventstore.Event) (*handle
 			handler.NewCol(LabelPolicyDarkLogoURLCol, nil),
 			handler.NewCol(LabelPolicyDarkIconURLCol, nil),
 		},
-		[]handler.Condition{
+		[]handler.Column{
-			handler.NewCond(LabelPolicyIDCol, event.Aggregate().ID),
+			handler.NewCol(LabelPolicyIDCol, event.Aggregate().ID),
-			handler.NewCond(LabelPolicyStateCol, domain.LabelPolicyStatePreview),
+			handler.NewCol(LabelPolicyStateCol, domain.LabelPolicyStatePreview),
-			handler.NewCond(LabelPolicyInstanceIDCol, event.Aggregate().InstanceID),
+			handler.NewCol(LabelPolicyInstanceIDCol, event.Aggregate().InstanceID),
 		}), nil
 }
 

internal/query/projection/milestones.go

@@ -248,17 +248,17 @@ func (p *milestoneProjection) reduceUserTokenAdded(event eventstore.Event) (*han
 	}
 	// We ignore authentications without app, for example JWT profile or PAT
 	if e.ApplicationID != "" {
-		crdb.AddUpdateStatement(
+		statements = append(statements, crdb.AddUpdateStatement(
 			[]handler.Column{
 				handler.NewCol(MilestoneColumnReachedDate, event.CreationDate()),
 			},
 			[]handler.Condition{
 				handler.NewCond(MilestoneColumnInstanceID, event.Aggregate().InstanceID),
 				handler.NewCond(MilestoneColumnType, milestone.AuthenticationSucceededOnApplication),
-				crdb.NewContainsCond(MilestoneColumnIgnoreClientIDs, e.ApplicationID),
+				crdb.Not(crdb.NewTextArrayContainsCond(MilestoneColumnIgnoreClientIDs, e.ApplicationID)),
 				crdb.NewIsNullCond(MilestoneColumnReachedDate),
 			},
-		)
+		))
 	}
 	return crdb.NewMultiStatement(e, statements...), nil
 }

internal/query/projection/projection.go

@@ -194,7 +194,7 @@ func applyCustomConfig(config crdb.StatementHandlerConfig, customConfig CustomCo
 // as setup and start currently create them individually, we make sure we get the right one
 // will be refactored when changing to new id based projections
 //
-// NotificationsProjection is not added here, because it does not statement based / has no proprietary projection table
+// Event handlers NotificationsProjection, NotificationsQuotaProjection and NotificationsProjection are not added here, because the do not statement based / have no proprietary projection table
 func newProjectionsList() {
 	projections = []projection{
 		OrgProjection,

internal/repository/milestone/events.go

@@ -14,6 +14,7 @@ const (
 type PushedEvent struct {
 	*eventstore.BaseEvent `json:"-"`
 	MilestoneType         Type     `json:"type"`
+	ExternalDomain        string   `json:"externalDomain"`
 	PrimaryDomain         string   `json:"primaryDomain"`
 	Endpoints             []string `json:"endpoints"`
 }
@@ -35,7 +36,7 @@ func NewPushedEvent(
 	aggregate *Aggregate,
 	msType Type,
 	endpoints []string,
-	primaryDomain string,
+	externalDomain, primaryDomain string,
 ) *PushedEvent {
 	return &PushedEvent{
 		BaseEvent: eventstore.NewBaseEventForPush(
@@ -43,8 +44,9 @@ func NewPushedEvent(
 			&aggregate.Aggregate,
 			PushedEventType,
 		),
-		MilestoneType: msType,
+		MilestoneType:  msType,
-		Endpoints:     endpoints,
+		Endpoints:      endpoints,
-		PrimaryDomain: primaryDomain,
+		ExternalDomain: externalDomain,
+		PrimaryDomain:  primaryDomain,
 	}
 }