feat: text query (#2735)

* feat: change mail template to new query side

* feat: adminapi message text

* feat: adminapi message text

* feat: adminapi message text

* feat: message texts

* feat: admin texts

* feat: tests

* feat: tests

* feat: custom login text on adminapi

* feat: custom login text

* feat: custom login text

* feat: message text prepare test

* feat: login text texts

* feat: custom login text

* merge main

* fix go.sum

Co-authored-by: Livio Amstutz <livio.a@gmail.com>

internal/notification/notification.go

@@ -2,10 +2,12 @@ package notification
 
 import (
 	"context"
+
 	"github.com/caos/logging"
 	"github.com/caos/zitadel/internal/command"
 	sd "github.com/caos/zitadel/internal/config/systemdefaults"
 	"github.com/caos/zitadel/internal/notification/repository/eventsourcing"
+	"github.com/caos/zitadel/internal/query"
 	"github.com/rakyll/statik/fs"
 
 	_ "github.com/caos/zitadel/internal/notification/statik"
@@ -16,7 +18,7 @@ type Config struct {
 	Repository eventsourcing.Config
 }
 
-func Start(ctx context.Context, config Config, systemDefaults sd.SystemDefaults, command *command.Commands, hasStatics bool) {
+func Start(ctx context.Context, config Config, systemDefaults sd.SystemDefaults, command *command.Commands, queries *query.Queries, hasStatics bool) {
 	statikFS, err := fs.NewWithNamespace("notification")
 	logging.Log("CONFI-7usEW").OnError(err).Panic("unable to start listener")
 
@@ -24,6 +26,6 @@ func Start(ctx context.Context, config Config, systemDefaults sd.SystemDefaults,
 	if !hasStatics {
 		apiDomain = ""
 	}
-	_, err = eventsourcing.Start(config.Repository, statikFS, systemDefaults, command, apiDomain)
+	_, err = eventsourcing.Start(config.Repository, statikFS, systemDefaults, command, queries, apiDomain)
 	logging.Log("MAIN-9uBxp").OnError(err).Panic("unable to start app")
 }

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

@@ -1,133 +0,0 @@
-package handler
-
-import (
-	"github.com/caos/logging"
-
-	caos_errs "github.com/caos/zitadel/internal/errors"
-	"github.com/caos/zitadel/internal/eventstore/v1"
-	es_models "github.com/caos/zitadel/internal/eventstore/v1/models"
-	"github.com/caos/zitadel/internal/eventstore/v1/query"
-	"github.com/caos/zitadel/internal/eventstore/v1/spooler"
-	iam_es_model "github.com/caos/zitadel/internal/iam/repository/eventsourcing/model"
-	iam_model "github.com/caos/zitadel/internal/iam/repository/view/model"
-	"github.com/caos/zitadel/internal/org/repository/eventsourcing/model"
-)
-
-type CustomText struct {
-	handler
-	subscription *v1.Subscription
-}
-
-func newCustomText(handler handler) *CustomText {
-	h := &CustomText{
-		handler: handler,
-	}
-
-	h.subscribe()
-
-	return h
-}
-
-func (m *CustomText) subscribe() {
-	m.subscription = m.es.Subscribe(m.AggregateTypes()...)
-	go func() {
-		for event := range m.subscription.Events {
-			query.ReduceEvent(m, event)
-		}
-	}()
-}
-
-const (
-	customTextTable = "notification.custom_texts"
-)
-
-func (m *CustomText) ViewModel() string {
-	return customTextTable
-}
-
-func (m *CustomText) Subscription() *v1.Subscription {
-	return m.subscription
-}
-
-func (_ *CustomText) AggregateTypes() []es_models.AggregateType {
-	return []es_models.AggregateType{model.OrgAggregate, iam_es_model.IAMAggregate}
-}
-
-func (p *CustomText) CurrentSequence() (uint64, error) {
-	sequence, err := p.view.GetLatestCustomTextSequence()
-	if err != nil {
-		return 0, err
-	}
-	return sequence.CurrentSequence, nil
-}
-
-func (m *CustomText) EventQuery() (*es_models.SearchQuery, error) {
-	sequence, err := m.view.GetLatestCustomTextSequence()
-	if err != nil {
-		return nil, err
-	}
-	return es_models.NewSearchQuery().
-		AggregateTypeFilter(m.AggregateTypes()...).
-		LatestSequenceFilter(sequence.CurrentSequence), nil
-}
-
-func (m *CustomText) Reduce(event *es_models.Event) (err error) {
-	switch event.AggregateType {
-	case model.OrgAggregate, iam_es_model.IAMAggregate:
-		err = m.processCustomText(event)
-	}
-	return err
-}
-
-func (m *CustomText) processCustomText(event *es_models.Event) (err error) {
-	customText := new(iam_model.CustomTextView)
-	switch event.Type {
-	case iam_es_model.CustomTextSet, model.CustomTextSet:
-		text := new(iam_model.CustomTextView)
-		err = text.SetData(event)
-		if err != nil {
-			return err
-		}
-		customText, err = m.view.CustomTextByIDs(event.AggregateID, text.Template, text.Key, text.Language)
-		if err != nil && !caos_errs.IsNotFound(err) {
-			return err
-		}
-		if caos_errs.IsNotFound(err) {
-			err = nil
-			customText = new(iam_model.CustomTextView)
-			customText.Language = text.Language
-			customText.Template = text.Template
-			customText.CreationDate = event.CreationDate
-		}
-		err = customText.AppendEvent(event)
-	case iam_es_model.CustomTextRemoved, model.CustomTextRemoved:
-		text := new(iam_model.CustomTextView)
-		err = text.SetData(event)
-		if err != nil {
-			return err
-		}
-		return m.view.DeleteCustomText(event.AggregateID, text.Template, text.Language, text.Key, event)
-	case iam_es_model.CustomTextMessageRemoved, model.CustomTextMessageRemoved:
-		text := new(iam_model.CustomTextView)
-		err = text.SetData(event)
-		if err != nil {
-			return err
-		}
-		return m.view.DeleteCustomTextTemplate(event.AggregateID, text.Template, text.Language, event)
-	default:
-		return m.view.ProcessedCustomTextSequence(event)
-	}
-	if err != nil {
-		return err
-	}
-	return m.view.PutCustomText(customText, event)
-}
-
-func (m *CustomText) OnError(event *es_models.Event, err error) error {
-	logging.LogWithFields("SPOOL-3m912", "id", event.AggregateID).WithError(err).Warn("something went wrong in custom text handler")
-	return spooler.HandleError(event, err, m.view.GetLatestCustomTextFailedEvent, m.view.ProcessedCustomTextFailedEvent, m.view.ProcessedCustomTextSequence, m.errorCountUntilSkip)
-}
-
-func (o *CustomText) OnSuccess() error {
-	return spooler.HandleSuccess(o.view.UpdateCustomTextSpoolerRunTimestamp)
-}

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

@@ -5,13 +5,14 @@ import (
 	"time"
 
 	"github.com/caos/logging"
+	"github.com/caos/zitadel/internal/query"
 
 	"github.com/caos/zitadel/internal/command"
 	sd "github.com/caos/zitadel/internal/config/systemdefaults"
 	"github.com/caos/zitadel/internal/config/types"
 	"github.com/caos/zitadel/internal/crypto"
 	v1 "github.com/caos/zitadel/internal/eventstore/v1"
-	"github.com/caos/zitadel/internal/eventstore/v1/query"
+	queryv1 "github.com/caos/zitadel/internal/eventstore/v1/query"
 	"github.com/caos/zitadel/internal/notification/repository/eventsourcing/view"
 )
 
@@ -34,20 +35,20 @@ func (h *handler) Eventstore() v1.Eventstore {
 	return h.es
 }
 
-func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, es v1.Eventstore, command *command.Commands, systemDefaults sd.SystemDefaults, dir http.FileSystem, apiDomain string) []query.Handler {
+func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, es v1.Eventstore, command *command.Commands, queries *query.Queries, systemDefaults sd.SystemDefaults, dir http.FileSystem, apiDomain string) []queryv1.Handler {
 	aesCrypto, err := crypto.NewAESCrypto(systemDefaults.UserVerificationKey)
 	if err != nil {
 		logging.Log("HANDL-s90ew").WithError(err).Debug("error create new aes crypto")
 	}
-	return []query.Handler{
+	return []queryv1.Handler{
 		newNotifyUser(
 			handler{view, bulkLimit, configs.cycleDuration("User"), errorCount, es},
 			systemDefaults.IamID,
 		),
-		newCustomText(handler{view, bulkLimit, configs.cycleDuration("CustomText"), errorCount, es}),
 		newNotification(
 			handler{view, bulkLimit, configs.cycleDuration("Notification"), errorCount, es},
 			command,
+			queries,
 			systemDefaults,
 			aesCrypto,
 			dir,

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

@@ -7,8 +7,6 @@ import (
 	"time"
 
 	"github.com/caos/logging"
-	"golang.org/x/text/language"
-
 	"github.com/caos/zitadel/internal/api/authz"
 	"github.com/caos/zitadel/internal/command"
 	sd "github.com/caos/zitadel/internal/config/systemdefaults"
@@ -17,12 +15,13 @@ import (
 	"github.com/caos/zitadel/internal/errors"
 	v1 "github.com/caos/zitadel/internal/eventstore/v1"
 	"github.com/caos/zitadel/internal/eventstore/v1/models"
-	"github.com/caos/zitadel/internal/eventstore/v1/query"
+	queryv1 "github.com/caos/zitadel/internal/eventstore/v1/query"
 	"github.com/caos/zitadel/internal/eventstore/v1/spooler"
 	"github.com/caos/zitadel/internal/i18n"
 	iam_model "github.com/caos/zitadel/internal/iam/model"
 	iam_es_model "github.com/caos/zitadel/internal/iam/repository/view/model"
 	"github.com/caos/zitadel/internal/notification/types"
+	"github.com/caos/zitadel/internal/query"
 	user_repo "github.com/caos/zitadel/internal/repository/user"
 	es_model "github.com/caos/zitadel/internal/user/repository/eventsourcing/model"
 	"github.com/caos/zitadel/internal/user/repository/view"
@@ -30,14 +29,10 @@ import (
 )
 
 const (
-	notificationTable    = "notification.notifications"
-	NotifyUserID         = "NOTIFICATION"
-	labelPolicyTableOrg  = "management.label_policies"
-	labelPolicyTableDef  = "adminapi.label_policies"
-	mailTemplateTableOrg = "management.mail_templates"
-	mailTemplateTableDef = "adminapi.mail_templates"
-	messageTextTableOrg  = "management.message_texts"
-	messageTextTableDef  = "adminapi.message_texts"
+	notificationTable   = "notification.notifications"
+	NotifyUserID        = "NOTIFICATION"
+	labelPolicyTableOrg = "management.label_policies"
+	labelPolicyTableDef = "adminapi.label_policies"
 )
 
 type Notification struct {
@@ -48,11 +43,13 @@ type Notification struct {
 	statikDir      http.FileSystem
 	subscription   *v1.Subscription
 	apiDomain      string
+	queries        *query.Queries
 }
 
 func newNotification(
 	handler handler,
 	command *command.Commands,
+	query *query.Queries,
 	defaults sd.SystemDefaults,
 	aesCrypto crypto.EncryptionAlgorithm,
 	statikDir http.FileSystem,
@@ -65,6 +62,7 @@ func newNotification(
 		statikDir:      statikDir,
 		AesCrypto:      aesCrypto,
 		apiDomain:      apiDomain,
+		queries:        query,
 	}
 
 	h.subscribe()
@@ -76,7 +74,7 @@ func (k *Notification) subscribe() {
 	k.subscription = k.es.Subscribe(k.AggregateTypes()...)
 	go func() {
 		for event := range k.subscription.Events {
-			query.ReduceEvent(k, event)
+			queryv1.ReduceEvent(k, event)
 		}
 	}()
 }
@@ -424,21 +422,20 @@ func (n *Notification) getLabelPolicy(ctx context.Context) (*iam_model.LabelPoli
 }
 
 // Read organization specific template
-func (n *Notification) getMailTemplate(ctx context.Context) (*iam_model.MailTemplateView, error) {
+func (n *Notification) getMailTemplate(ctx context.Context) (*query.MailTemplate, error) {
 	// read from Org
-	template, err := n.view.MailTemplateByAggregateID(authz.GetCtxData(ctx).OrgID, mailTemplateTableOrg)
+	template, err := n.queries.MailTemplateByOrg(ctx, authz.GetCtxData(ctx).OrgID)
 	if errors.IsNotFound(err) {
 		// read from default
-		template, err = n.view.MailTemplateByAggregateID(n.systemDefaults.IamID, mailTemplateTableDef)
+		template, err = n.queries.DefaultMailTemplate(ctx)
 		if err != nil {
 			return nil, err
 		}
-		template.Default = true
 	}
 	if err != nil {
 		return nil, err
 	}
-	return iam_es_model.MailTemplateViewToModel(template), err
+	return template, err
 }
 
 func (n *Notification) getTranslatorWithOrgTexts(orgID, textType string) (*i18n.Translator, error) {
@@ -446,52 +443,27 @@ func (n *Notification) getTranslatorWithOrgTexts(orgID, textType string) (*i18n.
 	if err != nil {
 		return nil, err
 	}
-	allCustomTexts, err := n.view.CustomTextsByAggregateIDAndTemplate(domain.IAMID, textType)
+	ctx := context.TODO()
+	allCustomTexts, err := n.queries.CustomTextListByTemplate(ctx, domain.IAMID, textType)
 	if err != nil {
 		return translator, nil
 	}
-	customTexts, err := n.view.CustomTextsByAggregateIDAndTemplate(orgID, textType)
+	customTexts, err := n.queries.CustomTextListByTemplate(ctx, orgID, textType)
 	if err != nil {
 		return translator, nil
 	}
-	allCustomTexts = append(allCustomTexts, customTexts...)
+	allCustomTexts.CustomTexts = append(allCustomTexts.CustomTexts, customTexts.CustomTexts...)
 
-	for _, text := range allCustomTexts {
+	for _, text := range allCustomTexts.CustomTexts {
 		msg := i18n.Message{
 			ID:   text.Template + "." + text.Key,
 			Text: text.Text,
 		}
-		translator.AddMessages(language.Make(text.Language), msg)
+		translator.AddMessages(text.Language, msg)
 	}
 	return translator, nil
 }
 
-// Read organization specific texts
-func (n *Notification) getMessageText(user *model.NotifyUser, textType, lang string) (*iam_model.MessageTextView, error) {
-	langTag := language.Make(lang)
-	if langTag == language.Und {
-		langTag = language.English
-	}
-	langBase, _ := langTag.Base()
-
-	defaultMessageText, err := n.view.MessageTextByIDs(n.systemDefaults.IamID, textType, langBase.String(), messageTextTableDef)
-	if err != nil {
-		return nil, err
-	}
-	defaultMessageText.Default = true
-
-	// read from Org
-	orgMessageText, err := n.view.MessageTextByIDs(user.ResourceOwner, textType, langBase.String(), messageTextTableOrg)
-	if errors.IsNotFound(err) {
-		return iam_es_model.MessageTextViewToModel(defaultMessageText), nil
-	}
-	if err != nil {
-		return nil, err
-	}
-	mergedText := mergeMessageTexts(defaultMessageText, orgMessageText)
-	return iam_es_model.MessageTextViewToModel(mergedText), err
-}
-
 func (n *Notification) getUserByID(userID string) (*model.NotifyUser, error) {
 	user, usrErr := n.view.NotifyUserByID(userID)
 	if usrErr != nil && !errors.IsNotFound(usrErr) {
@@ -515,28 +487,3 @@ func (n *Notification) getUserByID(userID string) (*model.NotifyUser, error) {
 	}
 	return &userCopy, nil
 }
-
-func mergeMessageTexts(defaultText *iam_es_model.MessageTextView, orgText *iam_es_model.MessageTextView) *iam_es_model.MessageTextView {
-	if orgText.Subject == "" {
-		orgText.Subject = defaultText.Subject
-	}
-	if orgText.Title == "" {
-		orgText.Title = defaultText.Title
-	}
-	if orgText.PreHeader == "" {
-		orgText.PreHeader = defaultText.PreHeader
-	}
-	if orgText.Text == "" {
-		orgText.Text = defaultText.Text
-	}
-	if orgText.Greeting == "" {
-		orgText.Greeting = defaultText.Greeting
-	}
-	if orgText.ButtonText == "" {
-		orgText.ButtonText = defaultText.ButtonText
-	}
-	if orgText.FooterText == "" {
-		orgText.FooterText = defaultText.FooterText
-	}
-	return orgText
-}

internal/notification/repository/eventsourcing/repository.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/caos/zitadel/internal/command"
 	"github.com/caos/zitadel/internal/eventstore/v1"
+	"github.com/caos/zitadel/internal/query"
 
 	"golang.org/x/text/language"
 
@@ -27,7 +28,7 @@ type EsRepository struct {
 	spooler *es_spol.Spooler
 }
 
-func Start(conf Config, dir http.FileSystem, systemDefaults sd.SystemDefaults, command *command.Commands, apiDomain string) (*EsRepository, error) {
+func Start(conf Config, dir http.FileSystem, systemDefaults sd.SystemDefaults, command *command.Commands, queries *query.Queries, apiDomain string) (*EsRepository, error) {
 	es, err := v1.Start(conf.Eventstore)
 	if err != nil {
 		return nil, err
@@ -42,7 +43,7 @@ func Start(conf Config, dir http.FileSystem, systemDefaults sd.SystemDefaults, c
 		return nil, err
 	}
 
-	spool := spooler.StartSpooler(conf.Spooler, es, view, sqlClient, command, systemDefaults, dir, apiDomain)
+	spool := spooler.StartSpooler(conf.Spooler, es, view, sqlClient, command, queries, systemDefaults, dir, apiDomain)
 
 	return &EsRepository{
 		spool,

internal/notification/repository/eventsourcing/spooler/spooler.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/caos/zitadel/internal/command"
 	"github.com/caos/zitadel/internal/eventstore/v1"
+	"github.com/caos/zitadel/internal/query"
 
 	sd "github.com/caos/zitadel/internal/config/systemdefaults"
 	"github.com/caos/zitadel/internal/eventstore/v1/spooler"
@@ -20,12 +21,12 @@ type SpoolerConfig struct {
 	Handlers              handler.Configs
 }
 
-func StartSpooler(c SpoolerConfig, es v1.Eventstore, view *view.View, sql *sql.DB, command *command.Commands, systemDefaults sd.SystemDefaults, dir http.FileSystem, apiDomain string) *spooler.Spooler {
+func StartSpooler(c SpoolerConfig, es v1.Eventstore, view *view.View, sql *sql.DB, command *command.Commands, queries *query.Queries, systemDefaults sd.SystemDefaults, dir http.FileSystem, apiDomain string) *spooler.Spooler {
 	spoolerConfig := spooler.Config{
 		Eventstore:        es,
 		Locker:            &locker{dbClient: sql},
 		ConcurrentWorkers: c.ConcurrentWorkers,
-		ViewHandlers:      handler.Register(c.Handlers, c.BulkLimit, c.FailureCountUntilSkip, view, es, command, systemDefaults, dir, apiDomain),
+		ViewHandlers:      handler.Register(c.Handlers, c.BulkLimit, c.FailureCountUntilSkip, view, es, command, queries, systemDefaults, dir, apiDomain),
 	}
 	spool := spoolerConfig.New()
 	spool.Start()

internal/notification/repository/eventsourcing/view/custom_texts.go

@@ -1,69 +0,0 @@
-package view
-
-import (
-	"github.com/caos/zitadel/internal/errors"
-	"github.com/caos/zitadel/internal/eventstore/v1/models"
-	"github.com/caos/zitadel/internal/iam/repository/view"
-	"github.com/caos/zitadel/internal/iam/repository/view/model"
-	global_view "github.com/caos/zitadel/internal/view/repository"
-)
-
-const (
-	customTextTable = "notification.custom_texts"
-)
-
-func (v *View) CustomTextByIDs(aggregateID, template, lang, key string) (*model.CustomTextView, error) {
-	return view.CustomTextByIDs(v.Db, customTextTable, aggregateID, template, lang, key)
-}
-
-func (v *View) CustomTextsByAggregateIDAndTemplate(aggregateID, template string) ([]*model.CustomTextView, error) {
-	return view.GetCustomTextsByAggregateIDAndTemplate(v.Db, customTextTable, aggregateID, template)
-}
-
-func (v *View) CustomTextsByAggregateIDAndTemplateAndLang(aggregateID, template, lang string) ([]*model.CustomTextView, error) {
-	return view.GetCustomTexts(v.Db, customTextTable, aggregateID, template, lang)
-}
-
-func (v *View) PutCustomText(template *model.CustomTextView, event *models.Event) error {
-	err := view.PutCustomText(v.Db, customTextTable, template)
-	if err != nil {
-		return err
-	}
-	return v.ProcessedCustomTextSequence(event)
-}
-
-func (v *View) DeleteCustomText(aggregateID, textType, lang, key string, event *models.Event) error {
-	err := view.DeleteCustomText(v.Db, customTextTable, aggregateID, textType, lang, key)
-	if err != nil && !errors.IsNotFound(err) {
-		return err
-	}
-	return v.ProcessedCustomTextSequence(event)
-}
-
-func (v *View) DeleteCustomTextTemplate(aggregateID, textType, lang string, event *models.Event) error {
-	err := view.DeleteCustomTextTemplate(v.Db, customTextTable, aggregateID, textType, lang)
-	if err != nil && !errors.IsNotFound(err) {
-		return err
-	}
-	return v.ProcessedCustomTextSequence(event)
-}
-
-func (v *View) GetLatestCustomTextSequence() (*global_view.CurrentSequence, error) {
-	return v.latestSequence(customTextTable)
-}
-
-func (v *View) ProcessedCustomTextSequence(event *models.Event) error {
-	return v.saveCurrentSequence(customTextTable, event)
-}
-
-func (v *View) UpdateCustomTextSpoolerRunTimestamp() error {
-	return v.updateSpoolerRunSequence(customTextTable)
-}
-
-func (v *View) GetLatestCustomTextFailedEvent(sequence uint64) (*global_view.FailedEvent, error) {
-	return v.latestFailedEvent(customTextTable, sequence)
-}
-
-func (v *View) ProcessedCustomTextFailedEvent(failedEvent *global_view.FailedEvent) error {
-	return v.saveFailedEvent(failedEvent)
-}

internal/notification/repository/eventsourcing/view/mail_template.go

@@ -1,10 +0,0 @@
-package view
-
-import (
-	"github.com/caos/zitadel/internal/iam/repository/view"
-	"github.com/caos/zitadel/internal/iam/repository/view/model"
-)
-
-func (v *View) MailTemplateByAggregateID(aggregateID string, mailTemplateTableVar string) (*model.MailTemplateView, error) {
-	return view.GetMailTemplateByAggregateID(v.Db, mailTemplateTableVar, aggregateID)
-}