zitadel/internal/notification/repository/eventsourcing/handler/notification.go

package handler

import (
	"context"
	"encoding/json"
	"net/http"
	"time"

	"github.com/zitadel/logging"

	"github.com/zitadel/zitadel/internal/api/authz"
	http_utils "github.com/zitadel/zitadel/internal/api/http"
	"github.com/zitadel/zitadel/internal/command"
	"github.com/zitadel/zitadel/internal/crypto"
	"github.com/zitadel/zitadel/internal/domain"
	"github.com/zitadel/zitadel/internal/errors"
	"github.com/zitadel/zitadel/internal/eventstore"
	v1 "github.com/zitadel/zitadel/internal/eventstore/v1"
	"github.com/zitadel/zitadel/internal/eventstore/v1/models"
	queryv1 "github.com/zitadel/zitadel/internal/eventstore/v1/query"
	"github.com/zitadel/zitadel/internal/eventstore/v1/spooler"
	"github.com/zitadel/zitadel/internal/i18n"
	"github.com/zitadel/zitadel/internal/notification/channels/fs"
	"github.com/zitadel/zitadel/internal/notification/channels/log"
	"github.com/zitadel/zitadel/internal/notification/channels/smtp"
	"github.com/zitadel/zitadel/internal/notification/channels/twilio"
	"github.com/zitadel/zitadel/internal/notification/types"
	"github.com/zitadel/zitadel/internal/query"
	user_repo "github.com/zitadel/zitadel/internal/repository/user"
	es_model "github.com/zitadel/zitadel/internal/user/repository/eventsourcing/model"
	"github.com/zitadel/zitadel/internal/user/repository/view"
	"github.com/zitadel/zitadel/internal/user/repository/view/model"
)

const (
	notificationTable = "notification.notifications"
	NotifyUserID      = "NOTIFICATION"
)

type Notification struct {
	handler
	command            *command.Commands
	fileSystemPath     string
	statikDir          http.FileSystem
	subscription       *v1.Subscription
	assetsPrefix       string
	queries            *query.Queries
	userDataCrypto     crypto.EncryptionAlgorithm
	smtpPasswordCrypto crypto.EncryptionAlgorithm
	smsTokenCrypto     crypto.EncryptionAlgorithm
	externalPort       uint16
	externalSecure     bool
}

func newNotification(
	handler handler,
	command *command.Commands,
	query *query.Queries,
	externalPort uint16,
	externalSecure bool,
	statikDir http.FileSystem,
	assetsPrefix,
	fileSystemPath string,
	userEncryption crypto.EncryptionAlgorithm,
	smtpEncryption crypto.EncryptionAlgorithm,
	smsEncryption crypto.EncryptionAlgorithm,
) *Notification {
	h := &Notification{
		handler:            handler,
		command:            command,
		statikDir:          statikDir,
		assetsPrefix:       assetsPrefix,
		queries:            query,
		userDataCrypto:     userEncryption,
		smtpPasswordCrypto: smtpEncryption,
		smsTokenCrypto:     smsEncryption,
		externalSecure:     externalSecure,
		externalPort:       externalPort,
		fileSystemPath:     fileSystemPath,
	}

	h.subscribe()

	return h
}

func (k *Notification) subscribe() {
	k.subscription = k.es.Subscribe(k.AggregateTypes()...)
	go func() {
		for event := range k.subscription.Events {
			queryv1.ReduceEvent(k, event)
		}
	}()
}

func (n *Notification) ViewModel() string {
	return notificationTable
}

func (n *Notification) Subscription() *v1.Subscription {
	return n.subscription
}

func (_ *Notification) AggregateTypes() []models.AggregateType {
	return []models.AggregateType{user_repo.AggregateType}
}

func (n *Notification) CurrentSequence(instanceID string) (uint64, error) {
	sequence, err := n.view.GetLatestNotificationSequence(instanceID)
	if err != nil {
		return 0, err
	}
	return sequence.CurrentSequence, nil
}

func (n *Notification) EventQuery() (*models.SearchQuery, error) {
	sequences, err := n.view.GetLatestNotificationSequences()
	if err != nil {
		return nil, err
	}
	query := models.NewSearchQuery()
	instances := make([]string, 0)
	for _, sequence := range sequences {
		for _, instance := range instances {
			if sequence.InstanceID == instance {
				break
			}
		}
		instances = append(instances, sequence.InstanceID)
		query.AddQuery().
			AggregateTypeFilter(n.AggregateTypes()...).
			LatestSequenceFilter(sequence.CurrentSequence).
			InstanceIDFilter(sequence.InstanceID)
	}
	return query.AddQuery().
		AggregateTypeFilter(n.AggregateTypes()...).
		LatestSequenceFilter(0).
		ExcludedInstanceIDsFilter(instances...).
		SearchQuery(), nil
}

func (n *Notification) Reduce(event *models.Event) (err error) {
	switch eventstore.EventType(event.Type) {
	case user_repo.UserV1InitialCodeAddedType,
		user_repo.HumanInitialCodeAddedType:
		err = n.handleInitUserCode(event)
	case user_repo.UserV1EmailCodeAddedType,
		user_repo.HumanEmailCodeAddedType:
		err = n.handleEmailVerificationCode(event)
	case user_repo.UserV1PhoneCodeAddedType,
		user_repo.HumanPhoneCodeAddedType:
		err = n.handlePhoneVerificationCode(event)
	case user_repo.UserV1PasswordCodeAddedType,
		user_repo.HumanPasswordCodeAddedType:
		err = n.handlePasswordCode(event)
	case user_repo.UserDomainClaimedType:
		err = n.handleDomainClaimed(event)
	case user_repo.HumanPasswordlessInitCodeRequestedType:
		err = n.handlePasswordlessRegistrationLink(event)
	}
	if err != nil {
		return err
	}
	return n.view.ProcessedNotificationSequence(event)
}

func (n *Notification) handleInitUserCode(event *models.Event) (err error) {
	initCode := new(es_model.InitUserCode)
	if err := initCode.SetData(event); err != nil {
		return err
	}
	ctx := getSetNotifyContextData(event.InstanceID, event.ResourceOwner)
	alreadyHandled, err := n.checkIfCodeAlreadyHandledOrExpired(ctx, event, initCode.Expiry,
		user_repo.UserV1InitialCodeAddedType, user_repo.UserV1InitialCodeSentType,
		user_repo.HumanInitialCodeAddedType, user_repo.HumanInitialCodeSentType)
	if err != nil || alreadyHandled {
		return err
	}
	colors, err := n.getLabelPolicy(ctx)
	if err != nil {
		return err
	}

	template, err := n.getMailTemplate(ctx)
	if err != nil {
		return err
	}

	user, err := n.getUserByID(event.AggregateID, event.InstanceID)
	if err != nil {
		return err
	}

	translator, err := n.getTranslatorWithOrgTexts(ctx, user.ResourceOwner, domain.InitCodeMessageType)
	if err != nil {
		return err
	}

	origin, err := n.origin(ctx)
	if err != nil {
		return err
	}
	err = types.SendUserInitCode(ctx, string(template.Template), translator, user, initCode, n.getSMTPConfig, n.getFileSystemProvider, n.getLogProvider, n.userDataCrypto, colors, n.assetsPrefix, origin)
	if err != nil {
		return err
	}
	return n.command.HumanInitCodeSent(ctx, event.ResourceOwner, event.AggregateID)
}

func (n *Notification) handlePasswordCode(event *models.Event) (err error) {
	pwCode := new(es_model.PasswordCode)
	if err := pwCode.SetData(event); err != nil {
		return err
	}
	ctx := getSetNotifyContextData(event.InstanceID, event.ResourceOwner)
	alreadyHandled, err := n.checkIfCodeAlreadyHandledOrExpired(ctx, event, pwCode.Expiry,
		user_repo.UserV1PasswordCodeAddedType, user_repo.UserV1PasswordCodeSentType,
		user_repo.HumanPasswordCodeAddedType, user_repo.HumanPasswordCodeSentType)
	if err != nil || alreadyHandled {
		return err
	}
	colors, err := n.getLabelPolicy(ctx)
	if err != nil {
		return err
	}

	template, err := n.getMailTemplate(ctx)
	if err != nil {
		return err
	}

	user, err := n.getUserByID(event.AggregateID, event.InstanceID)
	if err != nil {
		return err
	}

	translator, err := n.getTranslatorWithOrgTexts(ctx, user.ResourceOwner, domain.PasswordResetMessageType)
	if err != nil {
		return err
	}

	origin, err := n.origin(ctx)
	if err != nil {
		return err
	}
	err = types.SendPasswordCode(ctx, string(template.Template), translator, user, pwCode, n.getSMTPConfig, n.getTwilioConfig, n.getFileSystemProvider, n.getLogProvider, n.userDataCrypto, colors, n.assetsPrefix, origin)
	if err != nil {
		return err
	}
	return n.command.PasswordCodeSent(ctx, event.ResourceOwner, event.AggregateID)
}

func (n *Notification) handleEmailVerificationCode(event *models.Event) (err error) {
	emailCode := new(es_model.EmailCode)
	if err := emailCode.SetData(event); err != nil {
		return err
	}
	ctx := getSetNotifyContextData(event.InstanceID, event.ResourceOwner)
	alreadyHandled, err := n.checkIfCodeAlreadyHandledOrExpired(ctx, event, emailCode.Expiry,
		user_repo.UserV1EmailCodeAddedType, user_repo.UserV1EmailCodeSentType,
		user_repo.HumanEmailCodeAddedType, user_repo.HumanEmailCodeSentType)
	if err != nil || alreadyHandled {
		return nil
	}
	colors, err := n.getLabelPolicy(ctx)
	if err != nil {
		return err
	}

	template, err := n.getMailTemplate(ctx)
	if err != nil {
		return err
	}

	user, err := n.getUserByID(event.AggregateID, event.InstanceID)
	if err != nil {
		return err
	}

	translator, err := n.getTranslatorWithOrgTexts(ctx, user.ResourceOwner, domain.VerifyEmailMessageType)
	if err != nil {
		return err
	}

	origin, err := n.origin(ctx)
	if err != nil {
		return err
	}
	err = types.SendEmailVerificationCode(ctx, string(template.Template), translator, user, emailCode, n.getSMTPConfig, n.getFileSystemProvider, n.getLogProvider, n.userDataCrypto, colors, n.assetsPrefix, origin)
	if err != nil {
		return err
	}
	return n.command.HumanEmailVerificationCodeSent(ctx, event.ResourceOwner, event.AggregateID)
}

func (n *Notification) handlePhoneVerificationCode(event *models.Event) (err error) {
	phoneCode := new(es_model.PhoneCode)
	if err := phoneCode.SetData(event); err != nil {
		return err
	}
	ctx := getSetNotifyContextData(event.InstanceID, event.ResourceOwner)
	alreadyHandled, err := n.checkIfCodeAlreadyHandledOrExpired(ctx, event, phoneCode.Expiry,
		user_repo.UserV1PhoneCodeAddedType, user_repo.UserV1PhoneCodeSentType,
		user_repo.HumanPhoneCodeAddedType, user_repo.HumanPhoneCodeSentType)
	if err != nil || alreadyHandled {
		return nil
	}
	user, err := n.getUserByID(event.AggregateID, event.InstanceID)
	if err != nil {
		return err
	}
	translator, err := n.getTranslatorWithOrgTexts(ctx, user.ResourceOwner, domain.VerifyPhoneMessageType)
	if err != nil {
		return err
	}
	err = types.SendPhoneVerificationCode(ctx, translator, user, phoneCode, n.getTwilioConfig, n.getFileSystemProvider, n.getLogProvider, n.userDataCrypto)
	if err != nil {
		return err
	}
	return n.command.HumanPhoneVerificationCodeSent(ctx, event.ResourceOwner, event.AggregateID)
}

func (n *Notification) handleDomainClaimed(event *models.Event) (err error) {
	ctx := getSetNotifyContextData(event.InstanceID, event.ResourceOwner)
	alreadyHandled, err := n.checkIfAlreadyHandled(ctx, event.AggregateID, event.InstanceID, event.Sequence, user_repo.UserDomainClaimedType, user_repo.UserDomainClaimedSentType)
	if err != nil || alreadyHandled {
		return nil
	}
	data := make(map[string]string)
	if err := json.Unmarshal(event.Data, &data); err != nil {
		logging.Log("HANDLE-Gghq2").WithError(err).Error("could not unmarshal event data")
		return errors.ThrowInternal(err, "HANDLE-7hgj3", "could not unmarshal event")
	}
	user, err := n.getUserByID(event.AggregateID, event.InstanceID)
	if err != nil {
		return err
	}
	if user.LastEmail == "" {
		return nil
	}
	colors, err := n.getLabelPolicy(ctx)
	if err != nil {
		return err
	}

	template, err := n.getMailTemplate(ctx)
	if err != nil {
		return err
	}

	translator, err := n.getTranslatorWithOrgTexts(ctx, user.ResourceOwner, domain.DomainClaimedMessageType)
	if err != nil {
		return err
	}

	origin, err := n.origin(ctx)
	if err != nil {
		return err
	}
	err = types.SendDomainClaimed(ctx, string(template.Template), translator, user, data["userName"], n.getSMTPConfig, n.getFileSystemProvider, n.getLogProvider, colors, n.assetsPrefix, origin)
	if err != nil {
		return err
	}
	return n.command.UserDomainClaimedSent(ctx, event.ResourceOwner, event.AggregateID)
}

func (n *Notification) handlePasswordlessRegistrationLink(event *models.Event) (err error) {
	addedEvent := new(user_repo.HumanPasswordlessInitCodeRequestedEvent)
	if err := json.Unmarshal(event.Data, addedEvent); err != nil {
		return err
	}
	ctx := getSetNotifyContextData(event.InstanceID, event.ResourceOwner)
	events, err := n.getUserEvents(ctx, event.AggregateID, event.InstanceID, event.Sequence)
	if err != nil {
		return err
	}
	for _, e := range events {
		if eventstore.EventType(e.Type) == user_repo.HumanPasswordlessInitCodeSentType {
			sentEvent := new(user_repo.HumanPasswordlessInitCodeSentEvent)
			if err := json.Unmarshal(e.Data, sentEvent); err != nil {
				return err
			}
			if sentEvent.ID == addedEvent.ID {
				return nil
			}
		}
	}
	user, err := n.getUserByID(event.AggregateID, event.InstanceID)
	if err != nil {
		return err
	}
	colors, err := n.getLabelPolicy(ctx)
	if err != nil {
		return err
	}

	template, err := n.getMailTemplate(ctx)
	if err != nil {
		return err
	}

	translator, err := n.getTranslatorWithOrgTexts(ctx, user.ResourceOwner, domain.PasswordlessRegistrationMessageType)
	if err != nil {
		return err
	}

	origin, err := n.origin(ctx)
	if err != nil {
		return err
	}
	err = types.SendPasswordlessRegistrationLink(ctx, string(template.Template), translator, user, addedEvent, n.getSMTPConfig, n.getFileSystemProvider, n.getLogProvider, n.userDataCrypto, colors, n.assetsPrefix, origin)
	if err != nil {
		return err
	}
	return n.command.HumanPasswordlessInitCodeSent(ctx, event.AggregateID, event.ResourceOwner, addedEvent.ID)
}

func (n *Notification) checkIfCodeAlreadyHandledOrExpired(ctx context.Context, event *models.Event, expiry time.Duration, eventTypes ...eventstore.EventType) (bool, error) {
	if event.CreationDate.Add(expiry).Before(time.Now().UTC()) {
		return true, nil
	}
	return n.checkIfAlreadyHandled(ctx, event.AggregateID, event.InstanceID, event.Sequence, eventTypes...)
}

func (n *Notification) checkIfAlreadyHandled(ctx context.Context, userID, instanceID string, sequence uint64, eventTypes ...eventstore.EventType) (bool, error) {
	events, err := n.getUserEvents(ctx, userID, instanceID, sequence)
	if err != nil {
		return false, err
	}
	for _, event := range events {
		for _, eventType := range eventTypes {
			if eventstore.EventType(event.Type) == eventType {
				return true, nil
			}
		}
	}
	return false, nil
}

func (n *Notification) getUserEvents(ctx context.Context, userID, instanceID string, sequence uint64) ([]*models.Event, error) {
	query, err := view.UserByIDQuery(userID, instanceID, sequence)
	if err != nil {
		return nil, err
	}

	return n.es.FilterEvents(ctx, query)
}

func (n *Notification) OnError(event *models.Event, err error) error {
	logging.WithFields("id", event.AggregateID, "sequence", event.Sequence).WithError(err).Warn("something went wrong in notification handler")
	return spooler.HandleError(event, err, n.view.GetLatestNotificationFailedEvent, n.view.ProcessedNotificationFailedEvent, n.view.ProcessedNotificationSequence, n.errorCountUntilSkip)
}

func (n *Notification) OnSuccess() error {
	return spooler.HandleSuccess(n.view.UpdateNotificationSpoolerRunTimestamp)
}

func getSetNotifyContextData(instanceID, orgID string) context.Context {
	ctx := authz.WithInstanceID(context.Background(), instanceID)
	return authz.SetCtxData(ctx, authz.CtxData{UserID: NotifyUserID, OrgID: orgID})
}

// Read organization specific colors
func (n *Notification) getLabelPolicy(ctx context.Context) (*query.LabelPolicy, error) {
	return n.queries.ActiveLabelPolicyByOrg(ctx, authz.GetCtxData(ctx).OrgID)
}

// Read organization specific template
func (n *Notification) getMailTemplate(ctx context.Context) (*query.MailTemplate, error) {
	return n.queries.MailTemplateByOrg(ctx, authz.GetCtxData(ctx).OrgID)
}

// Read iam smtp config
func (n *Notification) getSMTPConfig(ctx context.Context) (*smtp.EmailConfig, error) {
	config, err := n.queries.SMTPConfigByAggregateID(ctx, authz.GetInstance(ctx).InstanceID())
	if err != nil {
		return nil, err
	}
	password, err := crypto.Decrypt(config.Password, n.smtpPasswordCrypto)
	if err != nil {
		return nil, err
	}
	return &smtp.EmailConfig{
		From:     config.SenderAddress,
		FromName: config.SenderName,
		SMTP: smtp.SMTP{
			Host:     config.Host,
			User:     config.User,
			Password: string(password),
		},
	}, nil
}

// Read iam twilio config
func (n *Notification) getTwilioConfig(ctx context.Context) (*twilio.TwilioConfig, error) {
	config, err := n.queries.SMSProviderConfigByID(ctx, authz.GetInstance(ctx).InstanceID())
	if err != nil {
		return nil, err
	}
	if config.TwilioConfig == nil {
		return nil, errors.ThrowNotFound(nil, "HANDLER-8nfow", "Errors.SMS.Twilio.NotFound")
	}
	token, err := crypto.Decrypt(config.TwilioConfig.Token, n.smtpPasswordCrypto)
	if err != nil {
		return nil, err
	}
	return &twilio.TwilioConfig{
		SID:          config.TwilioConfig.SID,
		Token:        string(token),
		SenderNumber: config.TwilioConfig.SenderNumber,
	}, nil
}

// Read iam filesystem provider config
func (n *Notification) getFileSystemProvider(ctx context.Context) (*fs.FSConfig, error) {
	config, err := n.queries.NotificationProviderByIDAndType(ctx, authz.GetInstance(ctx).InstanceID(), domain.NotificationProviderTypeFile)
	if err != nil {
		return nil, err
	}
	return &fs.FSConfig{
		Compact: config.Compact,
		Path:    n.fileSystemPath,
	}, nil
}

// Read iam log provider config
func (n *Notification) getLogProvider(ctx context.Context) (*log.LogConfig, error) {
	config, err := n.queries.NotificationProviderByIDAndType(ctx, authz.GetInstance(ctx).InstanceID(), domain.NotificationProviderTypeLog)
	if err != nil {
		return nil, err
	}
	return &log.LogConfig{
		Compact: config.Compact,
	}, nil
}

func (n *Notification) getTranslatorWithOrgTexts(ctx context.Context, orgID, textType string) (*i18n.Translator, error) {
	translator, err := i18n.NewTranslator(n.statikDir, n.queries.GetDefaultLanguage(ctx), "")
	if err != nil {
		return nil, err
	}

	allCustomTexts, err := n.queries.CustomTextListByTemplate(ctx, authz.GetInstance(ctx).InstanceID(), textType)
	if err != nil {
		return translator, nil
	}
	customTexts, err := n.queries.CustomTextListByTemplate(ctx, orgID, textType)
	if err != nil {
		return translator, nil
	}
	allCustomTexts.CustomTexts = append(allCustomTexts.CustomTexts, customTexts.CustomTexts...)

	for _, text := range allCustomTexts.CustomTexts {
		msg := i18n.Message{
			ID:   text.Template + "." + text.Key,
			Text: text.Text,
		}
		translator.AddMessages(text.Language, msg)
	}
	return translator, nil
}

func (n *Notification) getUserByID(userID, instanceID string) (*model.NotifyUser, error) {
	return n.view.NotifyUserByID(userID, instanceID)
}

func (n *Notification) origin(ctx context.Context) (string, error) {
	primary, err := query.NewInstanceDomainPrimarySearchQuery(true)
	domains, err := n.queries.SearchInstanceDomains(ctx, &query.InstanceDomainSearchQueries{
		Queries: []query.SearchQuery{primary},
	})
	if err != nil {
		return "", err
	}
	if len(domains.Domains) < 1 {
		return "", errors.ThrowInternal(nil, "NOTIF-Ef3r1", "Errors.Notification.NoDomain")
	}
	return http_utils.BuildHTTP(domains.Domains[0].Domain, n.externalPort, n.externalSecure), nil
}