feat: iam members in admin api (#272)

* feat: iam members in admin api

* feat: add error id in translate error

* fix: resolve merge conflicts

internal/admin/repository/eventsourcing/eventstore/iam.go

@@ -0,0 +1,67 @@
+package eventstore
+
+import (
+	"context"
+	admin_view "github.com/caos/zitadel/internal/admin/repository/eventsourcing/view"
+	"github.com/caos/zitadel/internal/config/systemdefaults"
+	iam_es_model "github.com/caos/zitadel/internal/iam/repository/view/model"
+	"strings"
+
+	iam_model "github.com/caos/zitadel/internal/iam/model"
+	iam_es "github.com/caos/zitadel/internal/iam/repository/eventsourcing"
+)
+
+type IamRepository struct {
+	SearchLimit uint64
+	*iam_es.IamEventstore
+	View           *admin_view.View
+	SystemDefaults systemdefaults.SystemDefaults
+	Roles          []string
+}
+
+func (repo *IamRepository) IamMemberByID(ctx context.Context, orgID, userID string) (*iam_model.IamMemberView, error) {
+	member, err := repo.View.IamMemberByIDs(orgID, userID)
+	if err != nil {
+		return nil, err
+	}
+	return iam_es_model.IamMemberToModel(member), nil
+}
+
+func (repo *IamRepository) AddIamMember(ctx context.Context, member *iam_model.IamMember) (*iam_model.IamMember, error) {
+	member.AggregateID = repo.SystemDefaults.IamID
+	return repo.IamEventstore.AddIamMember(ctx, member)
+}
+
+func (repo *IamRepository) ChangeIamMember(ctx context.Context, member *iam_model.IamMember) (*iam_model.IamMember, error) {
+	member.AggregateID = repo.SystemDefaults.IamID
+	return repo.IamEventstore.ChangeIamMember(ctx, member)
+}
+
+func (repo *IamRepository) RemoveIamMember(ctx context.Context, userID string) error {
+	member := iam_model.NewIamMember(repo.SystemDefaults.IamID, userID)
+	return repo.IamEventstore.RemoveIamMember(ctx, member)
+}
+
+func (repo *IamRepository) SearchIamMembers(ctx context.Context, request *iam_model.IamMemberSearchRequest) (*iam_model.IamMemberSearchResponse, error) {
+	request.EnsureLimit(repo.SearchLimit)
+	members, count, err := repo.View.SearchIamMembers(request)
+	if err != nil {
+		return nil, err
+	}
+	return &iam_model.IamMemberSearchResponse{
+		Offset:      request.Offset,
+		Limit:       request.Limit,
+		TotalResult: uint64(count),
+		Result:      iam_es_model.IamMembersToModel(members),
+	}, nil
+}
+
+func (repo *IamRepository) GetIamMemberRoles() []string {
+	roles := make([]string, 0)
+	for _, roleMap := range repo.Roles {
+		if strings.HasPrefix(roleMap, "IAM") {
+			roles = append(roles, roleMap)
+		}
+	}
+	return roles
+}

internal/admin/repository/eventsourcing/handler/handler.go

@@ -6,7 +6,6 @@ import (
 	"github.com/caos/zitadel/internal/admin/repository/eventsourcing/view"
 	"github.com/caos/zitadel/internal/config/types"
 	"github.com/caos/zitadel/internal/eventstore/spooler"
-	proj_event "github.com/caos/zitadel/internal/project/repository/eventsourcing"
 	usr_event "github.com/caos/zitadel/internal/user/repository/eventsourcing"
 )
 
@@ -24,13 +23,13 @@ type handler struct {
 }
 
 type EventstoreRepos struct {
-	ProjectEvents *proj_event.ProjectEventstore
-	UserEvents    *usr_event.UserEventstore
+	UserEvents *usr_event.UserEventstore
 }
 
-func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View) []spooler.Handler {
+func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, repos EventstoreRepos) []spooler.Handler {
 	return []spooler.Handler{
 		&Org{handler: handler{view, bulkLimit, configs.cycleDuration("Org"), errorCount}},
+		&IamMember{handler: handler{view, bulkLimit, configs.cycleDuration("IamMember"), errorCount}, userEvents: repos.UserEvents},
 	}
 }
 

internal/admin/repository/eventsourcing/handler/iam_member.go

@@ -0,0 +1,127 @@
+package handler
+
+import (
+	"context"
+	"github.com/caos/zitadel/internal/iam/repository/eventsourcing/model"
+	iam_model "github.com/caos/zitadel/internal/iam/repository/view/model"
+	"time"
+
+	"github.com/caos/logging"
+	"github.com/caos/zitadel/internal/eventstore/models"
+	es_models "github.com/caos/zitadel/internal/eventstore/models"
+	"github.com/caos/zitadel/internal/eventstore/spooler"
+	usr_model "github.com/caos/zitadel/internal/user/model"
+	usr_event "github.com/caos/zitadel/internal/user/repository/eventsourcing"
+	usr_es_model "github.com/caos/zitadel/internal/user/repository/eventsourcing/model"
+)
+
+type IamMember struct {
+	handler
+	userEvents *usr_event.UserEventstore
+}
+
+const (
+	iamMemberTable = "admin_api.iam_members"
+)
+
+func (m *IamMember) MinimumCycleDuration() time.Duration { return m.cycleDuration }
+
+func (m *IamMember) ViewModel() string {
+	return iamMemberTable
+}
+
+func (m *IamMember) EventQuery() (*models.SearchQuery, error) {
+	sequence, err := m.view.GetLatestIamMemberSequence()
+	if err != nil {
+		return nil, err
+	}
+	return es_models.NewSearchQuery().
+		AggregateTypeFilter(model.IamAggregate, usr_es_model.UserAggregate).
+		LatestSequenceFilter(sequence), nil
+}
+
+func (m *IamMember) Reduce(event *models.Event) (err error) {
+	switch event.AggregateType {
+	case model.IamAggregate:
+		err = m.processIamMember(event)
+	case usr_es_model.UserAggregate:
+		err = m.processUser(event)
+	}
+	return err
+}
+
+func (m *IamMember) processIamMember(event *models.Event) (err error) {
+	member := new(iam_model.IamMemberView)
+	switch event.Type {
+	case model.IamMemberAdded:
+		member.AppendEvent(event)
+		m.fillData(member)
+	case model.IamMemberChanged:
+		err := member.SetData(event)
+		if err != nil {
+			return err
+		}
+		member, err = m.view.IamMemberByIDs(event.AggregateID, member.UserID)
+		if err != nil {
+			return err
+		}
+		member.AppendEvent(event)
+	case model.IamMemberRemoved:
+		err := member.SetData(event)
+		if err != nil {
+			return err
+		}
+		return m.view.DeleteIamMember(event.AggregateID, member.UserID, event.Sequence)
+	default:
+		return m.view.ProcessedIamMemberSequence(event.Sequence)
+	}
+	if err != nil {
+		return err
+	}
+	return m.view.PutIamMember(member, member.Sequence)
+}
+
+func (m *IamMember) processUser(event *models.Event) (err error) {
+	switch event.Type {
+	case usr_es_model.UserProfileChanged,
+		usr_es_model.UserEmailChanged:
+		members, err := m.view.IamMembersByUserID(event.AggregateID)
+		if err != nil {
+			return err
+		}
+		user, err := m.userEvents.UserByID(context.Background(), event.AggregateID)
+		if err != nil {
+			return err
+		}
+		for _, member := range members {
+			m.fillUserData(member, user)
+			err = m.view.PutIamMember(member, event.Sequence)
+			if err != nil {
+				return err
+			}
+		}
+	default:
+		return m.view.ProcessedIamMemberSequence(event.Sequence)
+	}
+	return nil
+}
+
+func (m *IamMember) fillData(member *iam_model.IamMemberView) (err error) {
+	user, err := m.userEvents.UserByID(context.Background(), member.UserID)
+	if err != nil {
+		return err
+	}
+	m.fillUserData(member, user)
+	return nil
+}
+
+func (m *IamMember) fillUserData(member *iam_model.IamMemberView, user *usr_model.User) {
+	member.UserName = user.UserName
+	member.FirstName = user.FirstName
+	member.LastName = user.LastName
+	member.Email = user.EmailAddress
+}
+func (m *IamMember) OnError(event *models.Event, err error) error {
+	logging.LogWithFields("SPOOL-Ld9ow", "id", event.AggregateID).WithError(err).Warn("something went wrong in iammember handler")
+	return spooler.HandleError(event, err, m.view.GetLatestIamMemberFailedEvent, m.view.ProcessedIamMemberFailedEvent, m.view.ProcessedIamMemberSequence, m.errorCountUntilSkip)
+}

internal/admin/repository/eventsourcing/repository.go

@@ -2,6 +2,7 @@ package eventsourcing
 
 import (
 	"context"
+	"github.com/caos/zitadel/internal/admin/repository/eventsourcing/handler"
 	es_policy "github.com/caos/zitadel/internal/policy/repository/eventsourcing"
 
 	"github.com/caos/logging"
@@ -30,10 +31,11 @@ type Config struct {
 type EsRepository struct {
 	spooler *es_spol.Spooler
 	eventstore.OrgRepo
+	eventstore.IamRepository
 	eventstore.AdministratorRepo
 }
 
-func Start(ctx context.Context, conf Config, systemDefaults sd.SystemDefaults) (*EsRepository, error) {
+func Start(ctx context.Context, conf Config, systemDefaults sd.SystemDefaults, roles []string) (*EsRepository, error) {
 	es, err := es_int.Start(conf.Eventstore)
 	if err != nil {
 		return nil, err
@@ -84,7 +86,7 @@ func Start(ctx context.Context, conf Config, systemDefaults sd.SystemDefaults) (
 	err = setup.StartSetup(systemDefaults, eventstoreRepos).Execute(ctx)
 	logging.Log("SERVE-djs3R").OnError(err).Panic("failed to execute setup")
 
-	spool := spooler.StartSpooler(conf.Spooler, es, view, sqlClient)
+	spool := spooler.StartSpooler(conf.Spooler, es, view, sqlClient, handler.EventstoreRepos{UserEvents: user})
 
 	return &EsRepository{
 		spooler: spool,
@@ -96,6 +98,13 @@ func Start(ctx context.Context, conf Config, systemDefaults sd.SystemDefaults) (
 			View:             view,
 			SearchLimit:      conf.SearchLimit,
 		},
+		IamRepository: eventstore.IamRepository{
+			IamEventstore:  iam,
+			View:           view,
+			SystemDefaults: systemDefaults,
+			SearchLimit:    conf.SearchLimit,
+			Roles:          roles,
+		},
 		AdministratorRepo: eventstore.AdministratorRepo{
 			View: view,
 		},

internal/admin/repository/eventsourcing/spooler/spooler.go

@@ -16,12 +16,12 @@ type SpoolerConfig struct {
 	Handlers              handler.Configs
 }
 
-func StartSpooler(c SpoolerConfig, es eventstore.Eventstore, view *view.View, sql *sql.DB) *spooler.Spooler {
+func StartSpooler(c SpoolerConfig, es eventstore.Eventstore, view *view.View, sql *sql.DB, repos handler.EventstoreRepos) *spooler.Spooler {
 	spoolerConfig := spooler.Config{
 		Eventstore:      es,
 		Locker:          &locker{dbClient: sql},
 		ConcurrentTasks: c.ConcurrentTasks,
-		ViewHandlers:    handler.Register(c.Handlers, c.BulkLimit, c.FailureCountUntilSkip, view),
+		ViewHandlers:    handler.Register(c.Handlers, c.BulkLimit, c.FailureCountUntilSkip, view, repos),
 	}
 	spool := spoolerConfig.New()
 	spool.Start()

internal/admin/repository/eventsourcing/view/iam_member.go

@@ -0,0 +1,56 @@
+package view
+
+import (
+	iam_model "github.com/caos/zitadel/internal/iam/model"
+	"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 (
+	iamMemberTable = "admin_api.iam_members"
+)
+
+func (v *View) IamMemberByIDs(orgID, userID string) (*model.IamMemberView, error) {
+	return view.IamMemberByIDs(v.Db, iamMemberTable, orgID, userID)
+}
+
+func (v *View) SearchIamMembers(request *iam_model.IamMemberSearchRequest) ([]*model.IamMemberView, int, error) {
+	return view.SearchIamMembers(v.Db, iamMemberTable, request)
+}
+
+func (v *View) IamMembersByUserID(userID string) ([]*model.IamMemberView, error) {
+	return view.IamMembersByUserID(v.Db, iamMemberTable, userID)
+}
+
+func (v *View) PutIamMember(org *model.IamMemberView, sequence uint64) error {
+	err := view.PutIamMember(v.Db, iamMemberTable, org)
+	if err != nil {
+		return err
+	}
+	return v.ProcessedIamMemberSequence(sequence)
+}
+
+func (v *View) DeleteIamMember(orgID, userID string, eventSequence uint64) error {
+	err := view.DeleteIamMember(v.Db, iamMemberTable, orgID, userID)
+	if err != nil {
+		return nil
+	}
+	return v.ProcessedIamMemberSequence(eventSequence)
+}
+
+func (v *View) GetLatestIamMemberSequence() (uint64, error) {
+	return v.latestSequence(iamMemberTable)
+}
+
+func (v *View) ProcessedIamMemberSequence(eventSequence uint64) error {
+	return v.saveCurrentSequence(iamMemberTable, eventSequence)
+}
+
+func (v *View) GetLatestIamMemberFailedEvent(sequence uint64) (*global_view.FailedEvent, error) {
+	return v.latestFailedEvent(iamMemberTable, sequence)
+}
+
+func (v *View) ProcessedIamMemberFailedEvent(failedEvent *global_view.FailedEvent) error {
+	return v.saveFailedEvent(failedEvent)
+}

internal/admin/repository/iam.go

@@ -0,0 +1,15 @@
+package repository
+
+import (
+	"context"
+	iam_model "github.com/caos/zitadel/internal/iam/model"
+)
+
+type IamRepository interface {
+	SearchIamMembers(ctx context.Context, request *iam_model.IamMemberSearchRequest) (*iam_model.IamMemberSearchResponse, error)
+	AddIamMember(ctx context.Context, member *iam_model.IamMember) (*iam_model.IamMember, error)
+	ChangeIamMember(ctx context.Context, member *iam_model.IamMember) (*iam_model.IamMember, error)
+	RemoveIamMember(ctx context.Context, userID string) error
+
+	GetIamMemberRoles() []string
+}

internal/admin/repository/repository.go

@@ -5,5 +5,6 @@ import "context"
 type Repository interface {
 	Health(ctx context.Context) error
 	OrgRepository
+	IamRepository
 	AdministratorRepository
 }