feat: User metadata (#2025)

* feat: user meta data events

* feat: user meta data set tests

* feat: user meta data tests

* feat: user meta data in protos

* feat: user meta data command api

* feat: user meta data query side

* feat: proto correct order, fix handlers

* feat: proto correct order

* feat: fixes of pr comments

* feat: fixes of pr comments

* feat: value as byte array

* feat: metadata feature

* Update internal/auth/repository/eventsourcing/handler/meta_data.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* Update internal/command/user_meta_data.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* Update proto/zitadel/metadata.proto

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* Update proto/zitadel/metadata.proto

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* fix: rename metadata files and table

* fix: rename meta data to metadat in protos

* Update internal/domain/metadata.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* fix: rename vars

* fix: rebiuld docs

* Update internal/iam/repository/view/metadata_view.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

Co-authored-by: Silvan <silvan.reusser@gmail.com>

internal/auth/repository/eventsourcing/eventstore/user.go

@@ -14,6 +14,7 @@ import (
 	"github.com/caos/zitadel/internal/errors"
 	"github.com/caos/zitadel/internal/eventstore/v1"
 	"github.com/caos/zitadel/internal/eventstore/v1/models"
+	iam_model "github.com/caos/zitadel/internal/iam/repository/view/model"
 	key_model "github.com/caos/zitadel/internal/key/model"
 	key_view_model "github.com/caos/zitadel/internal/key/repository/view/model"
 	"github.com/caos/zitadel/internal/telemetry/tracing"
@@ -296,3 +297,39 @@ func (r *UserRepo) getUserEvents(ctx context.Context, userID string, sequence ui
 	}
 	return r.Eventstore.FilterEvents(ctx, query)
 }
+
+func (repo *UserRepo) GetMyMetadataByKey(ctx context.Context, key string) (*domain.Metadata, error) {
+	ctxData := authz.GetCtxData(ctx)
+	data, err := repo.View.MetadataByKeyAndResourceOwner(ctxData.UserID, ctxData.ResourceOwner, key)
+	if err != nil {
+		return nil, err
+	}
+	return iam_model.MetadataViewToDomain(data), nil
+}
+
+func (repo *UserRepo) SearchMyMetadata(ctx context.Context, req *domain.MetadataSearchRequest) (*domain.MetadataSearchResponse, error) {
+	ctxData := authz.GetCtxData(ctx)
+	err := req.EnsureLimit(repo.SearchLimit)
+	if err != nil {
+		return nil, err
+	}
+	sequence, sequenceErr := repo.View.GetLatestUserSequence()
+	logging.Log("EVENT-N9fsd").OnError(sequenceErr).Warn("could not read latest user sequence")
+	req.AppendAggregateIDQuery(ctxData.UserID)
+	req.AppendResourceOwnerQuery(ctxData.ResourceOwner)
+	metadata, count, err := repo.View.SearchMetadata(req)
+	if err != nil {
+		return nil, err
+	}
+	result := &domain.MetadataSearchResponse{
+		Offset:      req.Offset,
+		Limit:       req.Limit,
+		TotalResult: count,
+		Result:      iam_model.MetadataViewsToDomain(metadata),
+	}
+	if sequenceErr == nil {
+		result.Sequence = sequence.CurrentSequence
+		result.Timestamp = sequence.LastSuccessfulSpoolerRun
+	}
+	return result, nil
+}

internal/auth/repository/eventsourcing/handler/handler.go

@@ -72,6 +72,7 @@ func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, es
 		newRefreshToken(handler{view, bulkLimit, configs.cycleDuration("RefreshToken"), errorCount, es}),
 		newPrivacyPolicy(handler{view, bulkLimit, configs.cycleDuration("PrivacyPolicy"), errorCount, es}),
 		newCustomText(handler{view, bulkLimit, configs.cycleDuration("CustomTexts"), errorCount, es}),
+		newMetadata(handler{view, bulkLimit, configs.cycleDuration("Metadata"), errorCount, es}),
 	}
 }
 

internal/auth/repository/eventsourcing/handler/metadata.go

@@ -0,0 +1,124 @@
+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_model "github.com/caos/zitadel/internal/iam/repository/view/model"
+	usr_model "github.com/caos/zitadel/internal/user/repository/eventsourcing/model"
+)
+
+type Metadata struct {
+	handler
+	subscription *v1.Subscription
+}
+
+func newMetadata(handler handler) *Metadata {
+	h := &Metadata{
+		handler: handler,
+	}
+
+	h.subscribe()
+
+	return h
+}
+
+func (m *Metadata) subscribe() {
+	m.subscription = m.es.Subscribe(m.AggregateTypes()...)
+	go func() {
+		for event := range m.subscription.Events {
+			query.ReduceEvent(m, event)
+		}
+	}()
+}
+
+const (
+	metadataTable = "auth.metadata"
+)
+
+func (m *Metadata) ViewModel() string {
+	return metadataTable
+}
+
+func (m *Metadata) Subscription() *v1.Subscription {
+	return m.subscription
+}
+
+func (_ *Metadata) AggregateTypes() []es_models.AggregateType {
+	return []es_models.AggregateType{usr_model.UserAggregate}
+}
+
+func (p *Metadata) CurrentSequence() (uint64, error) {
+	sequence, err := p.view.GetLatestMetadataSequence()
+	if err != nil {
+		return 0, err
+	}
+	return sequence.CurrentSequence, nil
+}
+
+func (m *Metadata) EventQuery() (*es_models.SearchQuery, error) {
+	sequence, err := m.view.GetLatestMetadataSequence()
+	if err != nil {
+		return nil, err
+	}
+	return es_models.NewSearchQuery().
+		AggregateTypeFilter(m.AggregateTypes()...).
+		LatestSequenceFilter(sequence.CurrentSequence), nil
+}
+
+func (m *Metadata) Reduce(event *es_models.Event) (err error) {
+	switch event.AggregateType {
+	case usr_model.UserAggregate:
+		err = m.processMetadata(event)
+	}
+	return err
+}
+
+func (m *Metadata) processMetadata(event *es_models.Event) (err error) {
+	metadata := new(iam_model.MetadataView)
+	switch event.Type {
+	case usr_model.UserMetadataSet:
+		err = metadata.SetData(event)
+		if err != nil {
+			return err
+		}
+		metadata, err = m.view.MetadataByKey(event.AggregateID, metadata.Key)
+		if err != nil && !caos_errs.IsNotFound(err) {
+			return err
+		}
+		if caos_errs.IsNotFound(err) {
+			err = nil
+			metadata = new(iam_model.MetadataView)
+			metadata.CreationDate = event.CreationDate
+		}
+		err = metadata.AppendEvent(event)
+	case usr_model.UserMetadataRemoved:
+		data := new(iam_model.MetadataView)
+		err = data.SetData(event)
+		if err != nil {
+			return err
+		}
+		return m.view.DeleteMetadata(event.AggregateID, data.Key, event)
+	case usr_model.UserRemoved:
+		return m.view.DeleteMetadataByAggregateID(event.AggregateID, event)
+	default:
+		return m.view.ProcessedMetadataSequence(event)
+	}
+	if err != nil {
+		return err
+	}
+	return m.view.PutMetadata(metadata, event)
+}
+
+func (m *Metadata) OnError(event *es_models.Event, err error) error {
+	logging.LogWithFields("SPOOL-miJJs", "id", event.AggregateID).WithError(err).Warn("something went wrong in custom text handler")
+	return spooler.HandleError(event, err, m.view.GetLatestMetadataFailedEvent, m.view.ProcessedMetadataFailedEvent, m.view.ProcessedMetadataSequence, m.errorCountUntilSkip)
+}
+
+func (m *Metadata) OnSuccess() error {
+	return spooler.HandleSuccess(m.view.UpdateMetadataSpoolerRunTimestamp)
+}

internal/auth/repository/eventsourcing/view/metadata.go

@@ -0,0 +1,73 @@
+package view
+
+import (
+	"github.com/caos/zitadel/internal/domain"
+	"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 (
+	metadataTable = "auth.metadata"
+)
+
+func (v *View) MetadataByKey(aggregateID, key string) (*model.MetadataView, error) {
+	return view.MetadataByKey(v.Db, metadataTable, aggregateID, key)
+}
+
+func (v *View) MetadataListByAggregateID(aggregateID string) ([]*model.MetadataView, error) {
+	return view.GetMetadataList(v.Db, metadataTable, aggregateID)
+}
+
+func (v *View) MetadataByKeyAndResourceOwner(aggregateID, resourceOwner, key string) (*model.MetadataView, error) {
+	return view.MetadataByKeyAndResourceOwner(v.Db, metadataTable, aggregateID, resourceOwner, key)
+}
+
+func (v *View) SearchMetadata(request *domain.MetadataSearchRequest) ([]*model.MetadataView, uint64, error) {
+	return view.SearchMetadata(v.Db, metadataTable, request)
+}
+
+func (v *View) PutMetadata(template *model.MetadataView, event *models.Event) error {
+	err := view.PutMetadata(v.Db, metadataTable, template)
+	if err != nil {
+		return err
+	}
+	return v.ProcessedMetadataSequence(event)
+}
+
+func (v *View) DeleteMetadata(aggregateID, key string, event *models.Event) error {
+	err := view.DeleteMetadata(v.Db, metadataTable, aggregateID, key)
+	if err != nil && !errors.IsNotFound(err) {
+		return err
+	}
+	return v.ProcessedMetadataSequence(event)
+}
+
+func (v *View) DeleteMetadataByAggregateID(aggregateID string, event *models.Event) error {
+	err := view.DeleteMetadataByAggregateID(v.Db, metadataTable, aggregateID)
+	if err != nil && !errors.IsNotFound(err) {
+		return err
+	}
+	return v.ProcessedMetadataSequence(event)
+}
+func (v *View) GetLatestMetadataSequence() (*global_view.CurrentSequence, error) {
+	return v.latestSequence(metadataTable)
+}
+
+func (v *View) ProcessedMetadataSequence(event *models.Event) error {
+	return v.saveCurrentSequence(metadataTable, event)
+}
+
+func (v *View) UpdateMetadataSpoolerRunTimestamp() error {
+	return v.updateSpoolerRunSequence(metadataTable)
+}
+
+func (v *View) GetLatestMetadataFailedEvent(sequence uint64) (*global_view.FailedEvent, error) {
+	return v.latestFailedEvent(metadataTable, sequence)
+}
+
+func (v *View) ProcessedMetadataFailedEvent(failedEvent *global_view.FailedEvent) error {
+	return v.saveFailedEvent(failedEvent)
+}