zitadel/internal/notification/handlers/user_notifier_legacy.go

package handlers

import (
	"context"
	"errors"
	"strings"
	"time"

	http_util "github.com/zitadel/zitadel/internal/api/http"
	"github.com/zitadel/zitadel/internal/api/ui/login"
	"github.com/zitadel/zitadel/internal/crypto"
	"github.com/zitadel/zitadel/internal/domain"
	"github.com/zitadel/zitadel/internal/eventstore"
	"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
	"github.com/zitadel/zitadel/internal/notification/channels"
	"github.com/zitadel/zitadel/internal/notification/senders"
	"github.com/zitadel/zitadel/internal/notification/types"
	"github.com/zitadel/zitadel/internal/query"
	"github.com/zitadel/zitadel/internal/repository/session"
	"github.com/zitadel/zitadel/internal/repository/user"
	"github.com/zitadel/zitadel/internal/zerrors"
)

type userNotifierLegacy struct {
	commands     Commands
	queries      *NotificationQueries
	channels     types.ChannelChains
	otpEmailTmpl string
}

func NewUserNotifierLegacy(
	ctx context.Context,
	config handler.Config,
	commands Commands,
	queries *NotificationQueries,
	channels types.ChannelChains,
	otpEmailTmpl string,
) *handler.Handler {
	return handler.NewHandler(ctx, &config, &userNotifierLegacy{
		commands:     commands,
		queries:      queries,
		otpEmailTmpl: otpEmailTmpl,
		channels:     channels,
	})
}

func (u *userNotifierLegacy) Name() string {
	return UserNotificationsProjectionTable
}

func (u *userNotifierLegacy) Reducers() []handler.AggregateReducer {
	return []handler.AggregateReducer{
		{
			Aggregate: user.AggregateType,
			EventReducers: []handler.EventReducer{
				{
					Event:  user.UserV1InitialCodeAddedType,
					Reduce: u.reduceInitCodeAdded,
				},
				{
					Event:  user.HumanInitialCodeAddedType,
					Reduce: u.reduceInitCodeAdded,
				},
				{
					Event:  user.UserV1EmailCodeAddedType,
					Reduce: u.reduceEmailCodeAdded,
				},
				{
					Event:  user.HumanEmailCodeAddedType,
					Reduce: u.reduceEmailCodeAdded,
				},
				{
					Event:  user.UserV1PasswordCodeAddedType,
					Reduce: u.reducePasswordCodeAdded,
				},
				{
					Event:  user.HumanPasswordCodeAddedType,
					Reduce: u.reducePasswordCodeAdded,
				},
				{
					Event:  user.UserDomainClaimedType,
					Reduce: u.reduceDomainClaimed,
				},
				{
					Event:  user.HumanPasswordlessInitCodeRequestedType,
					Reduce: u.reducePasswordlessCodeRequested,
				},
				{
					Event:  user.UserV1PhoneCodeAddedType,
					Reduce: u.reducePhoneCodeAdded,
				},
				{
					Event:  user.HumanPhoneCodeAddedType,
					Reduce: u.reducePhoneCodeAdded,
				},
				{
					Event:  user.HumanPasswordChangedType,
					Reduce: u.reducePasswordChanged,
				},
				{
					Event:  user.HumanOTPSMSCodeAddedType,
					Reduce: u.reduceOTPSMSCodeAdded,
				},
				{
					Event:  user.HumanOTPEmailCodeAddedType,
					Reduce: u.reduceOTPEmailCodeAdded,
				},
				{
					Event:  user.HumanInviteCodeAddedType,
					Reduce: u.reduceInviteCodeAdded,
				},
			},
		},
		{
			Aggregate: session.AggregateType,
			EventReducers: []handler.EventReducer{
				{
					Event:  session.OTPSMSChallengedType,
					Reduce: u.reduceSessionOTPSMSChallenged,
				},
				{
					Event:  session.OTPEmailChallengedType,
					Reduce: u.reduceSessionOTPEmailChallenged,
				},
			},
		},
	}
}

func (u *userNotifierLegacy) reduceInitCodeAdded(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*user.HumanInitialCodeAddedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-EFe2f", "reduce.wrong.event.type %s", user.HumanInitialCodeAddedType)
	}

	return handler.NewStatement(event, func(ctx context.Context, ex handler.Executer, projectionName string) error {
		ctx = HandlerContext(ctx, event.Aggregate())
		alreadyHandled, err := u.checkIfCodeAlreadyHandledOrExpired(ctx, event, e.Expiry, nil,
			user.UserV1InitialCodeAddedType, user.UserV1InitialCodeSentType,
			user.HumanInitialCodeAddedType, user.HumanInitialCodeSentType)
		if err != nil {
			return err
		}
		if alreadyHandled {
			return nil
		}
		code, err := crypto.DecryptString(e.Code, u.queries.UserDataCrypto)
		if err != nil {
			return err
		}
		colors, err := u.queries.ActiveLabelPolicyByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		template, err := u.queries.MailTemplateByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		notifyUser, err := u.queries.GetNotifyUserByID(ctx, true, e.Aggregate().ID)
		if err != nil {
			return err
		}
		translator, err := u.queries.GetTranslatorWithOrgTexts(ctx, notifyUser.ResourceOwner, domain.InitCodeMessageType)
		if err != nil {
			return err
		}

		ctx, err = u.queries.Origin(ctx, e)
		if err != nil {
			return err
		}
		err = types.SendEmail(ctx, u.channels, string(template.Template), translator, notifyUser, colors, e.Type()).
			SendUserInitCode(ctx, notifyUser, code, e.AuthRequestID)
		if err != nil {
			if errors.Is(err, &channels.CancelError{}) {
				// if the notification was canceled, we don't want to return the error, so there is no retry
				return nil
			}
			return err
		}
		return u.commands.HumanInitCodeSent(ctx, e.Aggregate().ResourceOwner, e.Aggregate().ID)
	}), nil
}

func (u *userNotifierLegacy) reduceEmailCodeAdded(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*user.HumanEmailCodeAddedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-SWf3g", "reduce.wrong.event.type %s", user.HumanEmailCodeAddedType)
	}

	if e.CodeReturned {
		return handler.NewNoOpStatement(e), nil
	}

	return handler.NewStatement(event, func(ctx context.Context, ex handler.Executer, projectionName string) error {
		ctx = HandlerContext(ctx, event.Aggregate())
		alreadyHandled, err := u.checkIfCodeAlreadyHandledOrExpired(ctx, event, e.Expiry, nil,
			user.UserV1EmailCodeAddedType, user.UserV1EmailCodeSentType,
			user.HumanEmailCodeAddedType, user.HumanEmailCodeSentType)
		if err != nil {
			return err
		}
		if alreadyHandled {
			return nil
		}
		code, err := crypto.DecryptString(e.Code, u.queries.UserDataCrypto)
		if err != nil {
			return err
		}
		colors, err := u.queries.ActiveLabelPolicyByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		template, err := u.queries.MailTemplateByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		notifyUser, err := u.queries.GetNotifyUserByID(ctx, true, e.Aggregate().ID)
		if err != nil {
			return err
		}
		translator, err := u.queries.GetTranslatorWithOrgTexts(ctx, notifyUser.ResourceOwner, domain.VerifyEmailMessageType)
		if err != nil {
			return err
		}

		ctx, err = u.queries.Origin(ctx, e)
		if err != nil {
			return err
		}
		err = types.SendEmail(ctx, u.channels, string(template.Template), translator, notifyUser, colors, event.Type()).
			SendEmailVerificationCode(ctx, notifyUser, code, e.URLTemplate, e.AuthRequestID)
		if err != nil {
			if errors.Is(err, &channels.CancelError{}) {
				// if the notification was canceled, we don't want to return the error, so there is no retry
				return nil
			}
			return err
		}
		return u.commands.HumanEmailVerificationCodeSent(ctx, e.Aggregate().ResourceOwner, e.Aggregate().ID)
	}), nil
}

func (u *userNotifierLegacy) reducePasswordCodeAdded(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*user.HumanPasswordCodeAddedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-Eeg3s", "reduce.wrong.event.type %s", user.HumanPasswordCodeAddedType)
	}
	if e.CodeReturned {
		return handler.NewNoOpStatement(e), nil
	}

	return handler.NewStatement(event, func(ctx context.Context, ex handler.Executer, projectionName string) error {
		ctx = HandlerContext(ctx, event.Aggregate())
		alreadyHandled, err := u.checkIfCodeAlreadyHandledOrExpired(ctx, event, e.Expiry, nil,
			user.UserV1PasswordCodeAddedType, user.UserV1PasswordCodeSentType,
			user.HumanPasswordCodeAddedType, user.HumanPasswordCodeSentType)
		if err != nil {
			return err
		}
		if alreadyHandled {
			return nil
		}
		var code string
		if e.Code != nil {
			code, err = crypto.DecryptString(e.Code, u.queries.UserDataCrypto)
			if err != nil {
				return err
			}
		}
		colors, err := u.queries.ActiveLabelPolicyByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		template, err := u.queries.MailTemplateByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		notifyUser, err := u.queries.GetNotifyUserByID(ctx, true, e.Aggregate().ID)
		if err != nil {
			return err
		}
		translator, err := u.queries.GetTranslatorWithOrgTexts(ctx, notifyUser.ResourceOwner, domain.PasswordResetMessageType)
		if err != nil {
			return err
		}

		ctx, err = u.queries.Origin(ctx, e)
		if err != nil {
			return err
		}
		generatorInfo := new(senders.CodeGeneratorInfo)
		notify := types.SendEmail(ctx, u.channels, string(template.Template), translator, notifyUser, colors, event.Type())
		if e.NotificationType == domain.NotificationTypeSms {
			notify = types.SendSMS(ctx, u.channels, translator, notifyUser, colors, e.Type(), e.Aggregate().InstanceID, e.ID, generatorInfo)
		}
		err = notify.SendPasswordCode(ctx, notifyUser, code, e.URLTemplate, e.AuthRequestID)
		if err != nil {
			if errors.Is(err, &channels.CancelError{}) {
				// if the notification was canceled, we don't want to return the error, so there is no retry
				return nil
			}
			return err
		}
		return u.commands.PasswordCodeSent(ctx, e.Aggregate().ResourceOwner, e.Aggregate().ID, generatorInfo)
	}), nil
}

func (u *userNotifierLegacy) reduceOTPSMSCodeAdded(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*user.HumanOTPSMSCodeAddedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-ASF3g", "reduce.wrong.event.type %s", user.HumanOTPSMSCodeAddedType)
	}
	return u.reduceOTPSMS(
		e,
		e.Code,
		e.Expiry,
		e.Aggregate().ID,
		e.Aggregate().ResourceOwner,
		u.commands.HumanOTPSMSCodeSent,
		user.HumanOTPSMSCodeAddedType,
		user.HumanOTPSMSCodeSentType,
	)
}

func (u *userNotifierLegacy) reduceSessionOTPSMSChallenged(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*session.OTPSMSChallengedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-Sk32L", "reduce.wrong.event.type %s", session.OTPSMSChallengedType)
	}
	if e.CodeReturned {
		return handler.NewNoOpStatement(e), nil
	}
	ctx := HandlerContext(context.Background(), event.Aggregate())
	s, err := u.queries.SessionByID(ctx, true, e.Aggregate().ID, "", nil)
	if err != nil {
		return nil, err
	}
	return u.reduceOTPSMS(
		e,
		e.Code,
		e.Expiry,
		s.UserFactor.UserID,
		s.UserFactor.ResourceOwner,
		u.commands.OTPSMSSent,
		session.OTPSMSChallengedType,
		session.OTPSMSSentType,
	)
}

func (u *userNotifierLegacy) reduceOTPSMS(
	event eventstore.Event,
	code *crypto.CryptoValue,
	expiry time.Duration,
	userID,
	resourceOwner string,
	sentCommand func(ctx context.Context, userID, resourceOwner string, generatorInfo *senders.CodeGeneratorInfo) (err error),
	eventTypes ...eventstore.EventType,
) (*handler.Statement, error) {
	ctx := HandlerContext(context.Background(), event.Aggregate())
	alreadyHandled, err := u.checkIfCodeAlreadyHandledOrExpired(ctx, event, expiry, nil, eventTypes...)
	if err != nil {
		return nil, err
	}
	if alreadyHandled {
		return handler.NewNoOpStatement(event), nil
	}
	var plainCode string
	if code != nil {
		plainCode, err = crypto.DecryptString(code, u.queries.UserDataCrypto)
		if err != nil {
			return nil, err
		}
	}
	colors, err := u.queries.ActiveLabelPolicyByOrg(ctx, resourceOwner, false)
	if err != nil {
		return nil, err
	}

	notifyUser, err := u.queries.GetNotifyUserByID(ctx, true, userID)
	if err != nil {
		return nil, err
	}
	translator, err := u.queries.GetTranslatorWithOrgTexts(ctx, notifyUser.ResourceOwner, domain.VerifySMSOTPMessageType)
	if err != nil {
		return nil, err
	}
	ctx, err = u.queries.Origin(ctx, event)
	if err != nil {
		return nil, err
	}
	generatorInfo := new(senders.CodeGeneratorInfo)
	notify := types.SendSMS(ctx, u.channels, translator, notifyUser, colors, event.Type(), event.Aggregate().InstanceID, event.Aggregate().ID, generatorInfo)
	err = notify.SendOTPSMSCode(ctx, plainCode, expiry)
	if err != nil {
		if errors.Is(err, &channels.CancelError{}) {
			// if the notification was canceled, we don't want to return the error, so there is no retry
			return handler.NewNoOpStatement(event), nil
		}
		return nil, err
	}
	err = sentCommand(ctx, event.Aggregate().ID, event.Aggregate().ResourceOwner, generatorInfo)
	if err != nil {
		return nil, err
	}
	return handler.NewNoOpStatement(event), nil
}

func (u *userNotifierLegacy) reduceOTPEmailCodeAdded(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*user.HumanOTPEmailCodeAddedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-JL3hw", "reduce.wrong.event.type %s", user.HumanOTPEmailCodeAddedType)
	}
	var authRequestID string
	if e.AuthRequestInfo != nil {
		authRequestID = e.AuthRequestInfo.ID
	}
	url := func(code, origin string, _ *query.NotifyUser) (string, error) {
		return login.OTPLink(origin, authRequestID, code, domain.MFATypeOTPEmail), nil
	}
	return u.reduceOTPEmail(
		e,
		e.Code,
		e.Expiry,
		e.Aggregate().ID,
		e.Aggregate().ResourceOwner,
		url,
		u.commands.HumanOTPEmailCodeSent,
		user.HumanOTPEmailCodeAddedType,
		user.HumanOTPEmailCodeSentType,
	)
}

func (u *userNotifierLegacy) reduceSessionOTPEmailChallenged(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*session.OTPEmailChallengedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-zbsgt", "reduce.wrong.event.type %s", session.OTPEmailChallengedType)
	}
	if e.ReturnCode {
		return handler.NewNoOpStatement(e), nil
	}
	ctx := HandlerContext(context.Background(), event.Aggregate())
	s, err := u.queries.SessionByID(ctx, true, e.Aggregate().ID, "", nil)
	if err != nil {
		return nil, err
	}
	url := func(code, origin string, user *query.NotifyUser) (string, error) {
		var buf strings.Builder
		urlTmpl := origin + u.otpEmailTmpl
		if e.URLTmpl != "" {
			urlTmpl = e.URLTmpl
		}
		if err := domain.RenderOTPEmailURLTemplate(&buf, urlTmpl, code, user.ID, user.PreferredLoginName, user.DisplayName, e.Aggregate().ID, user.PreferredLanguage); err != nil {
			return "", err
		}
		return buf.String(), nil
	}
	return u.reduceOTPEmail(
		e,
		e.Code,
		e.Expiry,
		s.UserFactor.UserID,
		s.UserFactor.ResourceOwner,
		url,
		u.commands.OTPEmailSent,
		user.HumanOTPEmailCodeAddedType,
		user.HumanOTPEmailCodeSentType,
	)
}

func (u *userNotifierLegacy) reduceOTPEmail(
	event eventstore.Event,
	code *crypto.CryptoValue,
	expiry time.Duration,
	userID,
	resourceOwner string,
	urlTmpl func(code, origin string, user *query.NotifyUser) (string, error),
	sentCommand func(ctx context.Context, userID string, resourceOwner string) (err error),
	eventTypes ...eventstore.EventType,
) (*handler.Statement, error) {
	ctx := HandlerContext(context.Background(), event.Aggregate())
	alreadyHandled, err := u.checkIfCodeAlreadyHandledOrExpired(ctx, event, expiry, nil, eventTypes...)
	if err != nil {
		return nil, err
	}
	if alreadyHandled {
		return handler.NewNoOpStatement(event), nil
	}
	plainCode, err := crypto.DecryptString(code, u.queries.UserDataCrypto)
	if err != nil {
		return nil, err
	}
	colors, err := u.queries.ActiveLabelPolicyByOrg(ctx, resourceOwner, false)
	if err != nil {
		return nil, err
	}

	template, err := u.queries.MailTemplateByOrg(ctx, resourceOwner, false)
	if err != nil {
		return nil, err
	}

	notifyUser, err := u.queries.GetNotifyUserByID(ctx, true, userID)
	if err != nil {
		return nil, err
	}
	translator, err := u.queries.GetTranslatorWithOrgTexts(ctx, resourceOwner, domain.VerifyEmailOTPMessageType)
	if err != nil {
		return nil, err
	}
	ctx, err = u.queries.Origin(ctx, event)
	if err != nil {
		return nil, err
	}
	url, err := urlTmpl(plainCode, http_util.DomainContext(ctx).Origin(), notifyUser)
	if err != nil {
		return nil, err
	}
	notify := types.SendEmail(ctx, u.channels, string(template.Template), translator, notifyUser, colors, event.Type())
	err = notify.SendOTPEmailCode(ctx, url, plainCode, expiry)
	if err != nil {
		if errors.Is(err, &channels.CancelError{}) {
			// if the notification was canceled, we don't want to return the error, so there is no retry
			return handler.NewNoOpStatement(event), nil
		}
		return nil, err
	}
	err = sentCommand(ctx, event.Aggregate().ID, event.Aggregate().ResourceOwner)
	if err != nil {
		return nil, err
	}
	return handler.NewNoOpStatement(event), nil
}

func (u *userNotifierLegacy) reduceDomainClaimed(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*user.DomainClaimedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-Drh5w", "reduce.wrong.event.type %s", user.UserDomainClaimedType)
	}
	return handler.NewStatement(event, func(ctx context.Context, ex handler.Executer, projectionName string) error {
		ctx = HandlerContext(ctx, event.Aggregate())
		alreadyHandled, err := u.queries.IsAlreadyHandled(ctx, event, nil,
			user.UserDomainClaimedType, user.UserDomainClaimedSentType)
		if err != nil {
			return err
		}
		if alreadyHandled {
			return nil
		}
		colors, err := u.queries.ActiveLabelPolicyByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		template, err := u.queries.MailTemplateByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		notifyUser, err := u.queries.GetNotifyUserByID(ctx, true, e.Aggregate().ID)
		if err != nil {
			return err
		}
		translator, err := u.queries.GetTranslatorWithOrgTexts(ctx, notifyUser.ResourceOwner, domain.DomainClaimedMessageType)
		if err != nil {
			return err
		}

		ctx, err = u.queries.Origin(ctx, e)
		if err != nil {
			return err
		}
		err = types.SendEmail(ctx, u.channels, string(template.Template), translator, notifyUser, colors, event.Type()).
			SendDomainClaimed(ctx, notifyUser, e.UserName)
		if err != nil {
			if errors.Is(err, &channels.CancelError{}) {
				// if the notification was canceled, we don't want to return the error, so there is no retry
				return nil
			}
			return err
		}
		return u.commands.UserDomainClaimedSent(ctx, e.Aggregate().ResourceOwner, e.Aggregate().ID)
	}), nil
}

func (u *userNotifierLegacy) reducePasswordlessCodeRequested(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*user.HumanPasswordlessInitCodeRequestedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-EDtjd", "reduce.wrong.event.type %s", user.HumanPasswordlessInitCodeAddedType)
	}
	if e.CodeReturned {
		return handler.NewNoOpStatement(e), nil
	}

	return handler.NewStatement(event, func(ctx context.Context, ex handler.Executer, projectionName string) error {
		ctx = HandlerContext(ctx, event.Aggregate())
		alreadyHandled, err := u.checkIfCodeAlreadyHandledOrExpired(ctx, event, e.Expiry, map[string]interface{}{"id": e.ID}, user.HumanPasswordlessInitCodeSentType)
		if err != nil {
			return err
		}
		if alreadyHandled {
			return nil
		}
		code, err := crypto.DecryptString(e.Code, u.queries.UserDataCrypto)
		if err != nil {
			return err
		}
		colors, err := u.queries.ActiveLabelPolicyByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		template, err := u.queries.MailTemplateByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		notifyUser, err := u.queries.GetNotifyUserByID(ctx, true, e.Aggregate().ID)
		if err != nil {
			return err
		}
		translator, err := u.queries.GetTranslatorWithOrgTexts(ctx, notifyUser.ResourceOwner, domain.PasswordlessRegistrationMessageType)
		if err != nil {
			return err
		}

		ctx, err = u.queries.Origin(ctx, e)
		if err != nil {
			return err
		}
		err = types.SendEmail(ctx, u.channels, string(template.Template), translator, notifyUser, colors, event.Type()).
			SendPasswordlessRegistrationLink(ctx, notifyUser, code, e.ID, e.URLTemplate)
		if err != nil {
			if errors.Is(err, &channels.CancelError{}) {
				// if the notification was canceled, we don't want to return the error, so there is no retry
				return nil
			}
			return err
		}
		return u.commands.HumanPasswordlessInitCodeSent(ctx, e.Aggregate().ID, e.Aggregate().ResourceOwner, e.ID)
	}), nil
}

func (u *userNotifierLegacy) reducePasswordChanged(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*user.HumanPasswordChangedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-Yko2z8", "reduce.wrong.event.type %s", user.HumanPasswordChangedType)
	}

	return handler.NewStatement(event, func(ctx context.Context, ex handler.Executer, projectionName string) error {
		ctx = HandlerContext(ctx, event.Aggregate())
		alreadyHandled, err := u.queries.IsAlreadyHandled(ctx, event, nil, user.HumanPasswordChangeSentType)
		if err != nil {
			return err
		}
		if alreadyHandled {
			return nil
		}

		notificationPolicy, err := u.queries.NotificationPolicyByOrg(ctx, true, e.Aggregate().ResourceOwner, false)
		if zerrors.IsNotFound(err) {
			return nil
		}
		if err != nil {
			return err
		}

		if !notificationPolicy.PasswordChange {
			return nil
		}

		colors, err := u.queries.ActiveLabelPolicyByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		template, err := u.queries.MailTemplateByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		notifyUser, err := u.queries.GetNotifyUserByID(ctx, true, e.Aggregate().ID)
		if err != nil {
			return err
		}
		translator, err := u.queries.GetTranslatorWithOrgTexts(ctx, notifyUser.ResourceOwner, domain.PasswordChangeMessageType)
		if err != nil {
			return err
		}
		ctx, err = u.queries.Origin(ctx, e)
		if err != nil {
			return err
		}
		err = types.SendEmail(ctx, u.channels, string(template.Template), translator, notifyUser, colors, event.Type()).
			SendPasswordChange(ctx, notifyUser)
		if err != nil {
			if errors.Is(err, &channels.CancelError{}) {
				// if the notification was canceled, we don't want to return the error, so there is no retry
				return nil
			}
			return err
		}
		return u.commands.PasswordChangeSent(ctx, e.Aggregate().ResourceOwner, e.Aggregate().ID)
	}), nil
}

func (u *userNotifierLegacy) reducePhoneCodeAdded(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*user.HumanPhoneCodeAddedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-He83g", "reduce.wrong.event.type %s", user.HumanPhoneCodeAddedType)
	}
	if e.CodeReturned {
		return handler.NewNoOpStatement(e), nil
	}

	return handler.NewStatement(event, func(ctx context.Context, ex handler.Executer, projectionName string) error {
		ctx = HandlerContext(ctx, event.Aggregate())
		alreadyHandled, err := u.checkIfCodeAlreadyHandledOrExpired(ctx, event, e.Expiry, nil,
			user.UserV1PhoneCodeAddedType, user.UserV1PhoneCodeSentType,
			user.HumanPhoneCodeAddedType, user.HumanPhoneCodeSentType)
		if err != nil {
			return err
		}
		if alreadyHandled {
			return nil
		}
		var code string
		if e.Code != nil {
			code, err = crypto.DecryptString(e.Code, u.queries.UserDataCrypto)
			if err != nil {
				return err
			}
		}
		colors, err := u.queries.ActiveLabelPolicyByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		notifyUser, err := u.queries.GetNotifyUserByID(ctx, true, e.Aggregate().ID)
		if err != nil {
			return err
		}
		translator, err := u.queries.GetTranslatorWithOrgTexts(ctx, notifyUser.ResourceOwner, domain.VerifyPhoneMessageType)
		if err != nil {
			return err
		}

		ctx, err = u.queries.Origin(ctx, e)
		if err != nil {
			return err
		}
		generatorInfo := new(senders.CodeGeneratorInfo)
		if err = types.SendSMS(ctx, u.channels, translator, notifyUser, colors, e.Type(), e.Aggregate().InstanceID, e.ID, generatorInfo).
			SendPhoneVerificationCode(ctx, code); err != nil {
			if errors.Is(err, &channels.CancelError{}) {
				// if the notification was canceled, we don't want to return the error, so there is no retry
				return nil
			}
			return err
		}
		return u.commands.HumanPhoneVerificationCodeSent(ctx, e.Aggregate().ResourceOwner, e.Aggregate().ID, generatorInfo)
	}), nil
}

func (u *userNotifierLegacy) reduceInviteCodeAdded(event eventstore.Event) (*handler.Statement, error) {
	e, ok := event.(*user.HumanInviteCodeAddedEvent)
	if !ok {
		return nil, zerrors.ThrowInvalidArgumentf(nil, "HANDL-Eeg3s", "reduce.wrong.event.type %s", user.HumanInviteCodeAddedType)
	}
	if e.CodeReturned {
		return handler.NewNoOpStatement(e), nil
	}

	return handler.NewStatement(event, func(ctx context.Context, ex handler.Executer, projectionName string) error {
		ctx = HandlerContext(ctx, event.Aggregate())
		alreadyHandled, err := u.checkIfCodeAlreadyHandledOrExpired(ctx, event, e.Expiry, nil,
			user.HumanInviteCodeAddedType, user.HumanInviteCodeSentType)
		if err != nil {
			return err
		}
		if alreadyHandled {
			return nil
		}
		code, err := crypto.DecryptString(e.Code, u.queries.UserDataCrypto)
		if err != nil {
			return err
		}
		colors, err := u.queries.ActiveLabelPolicyByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		template, err := u.queries.MailTemplateByOrg(ctx, e.Aggregate().ResourceOwner, false)
		if err != nil {
			return err
		}

		notifyUser, err := u.queries.GetNotifyUserByID(ctx, true, e.Aggregate().ID)
		if err != nil {
			return err
		}
		translator, err := u.queries.GetTranslatorWithOrgTexts(ctx, notifyUser.ResourceOwner, domain.InviteUserMessageType)
		if err != nil {
			return err
		}

		ctx, err = u.queries.Origin(ctx, e)
		if err != nil {
			return err
		}
		notify := types.SendEmail(ctx, u.channels, string(template.Template), translator, notifyUser, colors, event.Type())
		err = notify.SendInviteCode(ctx, notifyUser, code, e.ApplicationName, e.URLTemplate, e.AuthRequestID)
		if err != nil {
			if errors.Is(err, &channels.CancelError{}) {
				// if the notification was canceled, we don't want to return the error, so there is no retry
				return nil
			}
			return err
		}
		return u.commands.InviteCodeSent(ctx, e.Aggregate().ID, e.Aggregate().ResourceOwner)
	}), nil
}

func (u *userNotifierLegacy) checkIfCodeAlreadyHandledOrExpired(ctx context.Context, event eventstore.Event, expiry time.Duration, data map[string]interface{}, eventTypes ...eventstore.EventType) (bool, error) {
	if expiry > 0 && event.CreatedAt().Add(expiry).Before(time.Now().UTC()) {
		return true, nil
	}
	return u.queries.IsAlreadyHandled(ctx, event, data, eventTypes...)
}