refactor(changes): use queries.SearchEvents (#5388)

* refactor(changes): use `queries.SearchEvents`

---------

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

internal/api/grpc/admin/event.go

@@ -17,7 +17,7 @@ func (s *Server) ListEvents(ctx context.Context, in *admin_pb.ListEventsRequest)
 	if err != nil {
 		return nil, err
 	}
-	events, err := s.query.SearchEvents(ctx, filter)
+	events, err := s.query.SearchEvents(ctx, filter, s.auditLogRetention)
 	if err != nil {
 		return nil, err
 	}

internal/api/grpc/admin/server.go

@@ -2,6 +2,8 @@ package admin
 
 import (
 	"context"
+	"time"
+
 	"google.golang.org/grpc"
 
 	"github.com/zitadel/zitadel/internal/admin/repository"
@@ -24,13 +26,14 @@ var _ admin.AdminServiceServer = (*Server)(nil)
 
 type Server struct {
 	admin.UnimplementedAdminServiceServer
-	database        string
-	command         *command.Commands
-	query           *query.Queries
-	administrator   repository.AdministratorRepository
-	assetsAPIDomain func(context.Context) string
-	userCodeAlg     crypto.EncryptionAlgorithm
-	passwordHashAlg crypto.HashAlgorithm
+	database          string
+	command           *command.Commands
+	query             *query.Queries
+	administrator     repository.AdministratorRepository
+	assetsAPIDomain   func(context.Context) string
+	userCodeAlg       crypto.EncryptionAlgorithm
+	passwordHashAlg   crypto.HashAlgorithm
+	auditLogRetention time.Duration
 }
 
 type Config struct {
@@ -45,15 +48,17 @@ func CreateServer(
 	repo repository.Repository,
 	externalSecure bool,
 	userCodeAlg crypto.EncryptionAlgorithm,
+	auditLogRetention time.Duration,
 ) *Server {
 	return &Server{
-		database:        database,
-		command:         command,
-		query:           query,
-		administrator:   repo,
-		assetsAPIDomain: assets.AssetAPI(externalSecure),
-		userCodeAlg:     userCodeAlg,
-		passwordHashAlg: crypto.NewBCrypt(sd.SecretGenerators.PasswordSaltCost),
+		database:          database,
+		command:           command,
+		query:             query,
+		administrator:     repo,
+		assetsAPIDomain:   assets.AssetAPI(externalSecure),
+		userCodeAlg:       userCodeAlg,
+		passwordHashAlg:   crypto.NewBCrypt(sd.SecretGenerators.PasswordSaltCost),
+		auditLogRetention: auditLogRetention,
 	}
 }
 

internal/api/grpc/auth/user.go

@@ -11,8 +11,10 @@ import (
 	user_grpc "github.com/zitadel/zitadel/internal/api/grpc/user"
 	"github.com/zitadel/zitadel/internal/command"
 	"github.com/zitadel/zitadel/internal/domain"
+	"github.com/zitadel/zitadel/internal/eventstore"
 	"github.com/zitadel/zitadel/internal/eventstore/v1/models"
 	"github.com/zitadel/zitadel/internal/query"
+	"github.com/zitadel/zitadel/internal/repository/user"
 	auth_pb "github.com/zitadel/zitadel/pkg/grpc/auth"
 )
 
@@ -56,13 +58,38 @@ func (s *Server) RemoveMyUser(ctx context.Context, _ *auth_pb.RemoveMyUserReques
 }
 
 func (s *Server) ListMyUserChanges(ctx context.Context, req *auth_pb.ListMyUserChangesRequest) (*auth_pb.ListMyUserChangesResponse, error) {
-	sequence, limit, asc := change.ChangeQueryToQuery(req.Query)
-	changes, err := s.query.UserChanges(ctx, authz.GetCtxData(ctx).UserID, sequence, limit, asc, s.auditLogRetention)
+	var (
+		limit    uint64
+		sequence uint64
+		asc      bool
+	)
+	if req.Query != nil {
+		limit = uint64(req.Query.Limit)
+		sequence = req.Query.Sequence
+		asc = req.Query.Asc
+	}
+
+	query := eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
+		AllowTimeTravel().
+		Limit(limit).
+		OrderDesc().
+		ResourceOwner(authz.GetCtxData(ctx).ResourceOwner).
+		AddQuery().
+		SequenceGreater(sequence).
+		AggregateTypes(user.AggregateType).
+		AggregateIDs(authz.GetCtxData(ctx).UserID).
+		Builder()
+	if asc {
+		query.OrderAsc()
+	}
+
+	changes, err := s.query.SearchEvents(ctx, query, s.auditLogRetention)
 	if err != nil {
 		return nil, err
 	}
+
 	return &auth_pb.ListMyUserChangesResponse{
-		Result: change.ChangesToPb(changes.Changes, s.assetsAPIDomain(ctx)),
+		Result: change.EventsToChangesPb(changes, s.assetsAPIDomain(ctx)),
 	}, nil
 }
 

internal/api/grpc/change/changes.go

@@ -10,30 +10,23 @@ import (
 	"github.com/zitadel/zitadel/pkg/grpc/message"
 )
 
-func ChangeQueryToQuery(query *change_pb.ChangeQuery) (sequence uint64, limit uint64, asc bool) {
-	if query == nil {
-		return 0, 0, false
-	}
-	return query.Sequence, uint64(query.Limit), query.Asc
-}
-
-func ChangesToPb(changes []*query.Change, assetAPIPrefix string) []*change_pb.Change {
+func EventsToChangesPb(changes []*query.Event, assetAPIPrefix string) []*change_pb.Change {
 	c := make([]*change_pb.Change, len(changes))
 	for i, change := range changes {
-		c[i] = ChangeToPb(change, assetAPIPrefix)
+		c[i] = EventToChangePb(change, assetAPIPrefix)
 	}
 	return c
 }
 
-func ChangeToPb(change *query.Change, assetAPIPrefix string) *change_pb.Change {
+func EventToChangePb(change *query.Event, assetAPIPrefix string) *change_pb.Change {
 	return &change_pb.Change{
-		ChangeDate:               timestamppb.New(change.ChangeDate),
-		EventType:                message.NewLocalizedEventType(change.EventType),
+		ChangeDate:               timestamppb.New(change.CreationDate),
+		EventType:                message.NewLocalizedEventType(change.Type),
 		Sequence:                 change.Sequence,
-		EditorId:                 change.ModifierId,
-		EditorDisplayName:        change.ModifierName,
-		EditorPreferredLoginName: change.ModifierLoginName,
-		EditorAvatarUrl:          domain.AvatarURL(assetAPIPrefix, change.ModifierResourceOwner, change.ModifierAvatarKey),
-		ResourceOwnerId:          change.ResourceOwner,
+		EditorId:                 change.Editor.ID,
+		EditorDisplayName:        change.Editor.DisplayName,
+		EditorPreferredLoginName: change.Editor.PreferedLoginName,
+		EditorAvatarUrl:          domain.AvatarURL(assetAPIPrefix, change.Aggregate.ResourceOwner, change.Editor.AvatarKey),
+		ResourceOwnerId:          change.Aggregate.ResourceOwner,
 	}
 }

internal/api/grpc/management/org.go

@@ -12,8 +12,10 @@ import (
 	org_grpc "github.com/zitadel/zitadel/internal/api/grpc/org"
 	policy_grpc "github.com/zitadel/zitadel/internal/api/grpc/policy"
 	"github.com/zitadel/zitadel/internal/domain"
+	"github.com/zitadel/zitadel/internal/eventstore"
 	"github.com/zitadel/zitadel/internal/eventstore/v1/models"
 	"github.com/zitadel/zitadel/internal/query"
+	"github.com/zitadel/zitadel/internal/repository/org"
 	mgmt_pb "github.com/zitadel/zitadel/pkg/grpc/management"
 )
 
@@ -35,13 +37,37 @@ func (s *Server) GetOrgByDomainGlobal(ctx context.Context, req *mgmt_pb.GetOrgBy
 }
 
 func (s *Server) ListOrgChanges(ctx context.Context, req *mgmt_pb.ListOrgChangesRequest) (*mgmt_pb.ListOrgChangesResponse, error) {
-	sequence, limit, asc := change_grpc.ChangeQueryToQuery(req.Query)
-	response, err := s.query.OrgChanges(ctx, authz.GetCtxData(ctx).OrgID, sequence, limit, asc, s.auditLogRetention)
+	var (
+		limit    uint64
+		sequence uint64
+		asc      bool
+	)
+	if req.Query != nil {
+		limit = uint64(req.Query.Limit)
+		sequence = req.Query.Sequence
+		asc = req.Query.Asc
+	}
+
+	query := eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
+		AllowTimeTravel().
+		Limit(limit).
+		OrderDesc().
+		ResourceOwner(authz.GetCtxData(ctx).OrgID).
+		AddQuery().
+		SequenceGreater(sequence).
+		AggregateTypes(org.AggregateType).
+		AggregateIDs(authz.GetCtxData(ctx).OrgID).
+		Builder()
+	if asc {
+		query.OrderAsc()
+	}
+
+	response, err := s.query.SearchEvents(ctx, query, s.auditLogRetention)
 	if err != nil {
 		return nil, err
 	}
 	return &mgmt_pb.ListOrgChangesResponse{
-		Result: change_grpc.ChangesToPb(response.Changes, s.assetAPIPrefix(ctx)),
+		Result: change_grpc.EventsToChangesPb(response, s.assetAPIPrefix(ctx)),
 	}, nil
 }
 

internal/api/grpc/management/project.go

@@ -8,7 +8,9 @@ import (
 	member_grpc "github.com/zitadel/zitadel/internal/api/grpc/member"
 	object_grpc "github.com/zitadel/zitadel/internal/api/grpc/object"
 	project_grpc "github.com/zitadel/zitadel/internal/api/grpc/project"
+	"github.com/zitadel/zitadel/internal/eventstore"
 	"github.com/zitadel/zitadel/internal/query"
+	"github.com/zitadel/zitadel/internal/repository/project"
 	mgmt_pb "github.com/zitadel/zitadel/pkg/grpc/management"
 )
 
@@ -56,13 +58,41 @@ func (s *Server) ListProjects(ctx context.Context, req *mgmt_pb.ListProjectsRequ
 }
 
 func (s *Server) ListProjectGrantChanges(ctx context.Context, req *mgmt_pb.ListProjectGrantChangesRequest) (*mgmt_pb.ListProjectGrantChangesResponse, error) {
-	sequence, limit, asc := change_grpc.ChangeQueryToQuery(req.Query)
-	res, err := s.query.ProjectGrantChanges(ctx, req.ProjectId, req.GrantId, sequence, limit, asc, s.auditLogRetention)
+	var (
+		limit    uint64
+		sequence uint64
+		asc      bool
+	)
+	if req.Query != nil {
+		limit = uint64(req.Query.Limit)
+		sequence = req.Query.Sequence
+		asc = req.Query.Asc
+	}
+
+	query := eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
+		AllowTimeTravel().
+		Limit(limit).
+		OrderDesc().
+		ResourceOwner(authz.GetCtxData(ctx).OrgID).
+		AddQuery().
+		SequenceGreater(sequence).
+		AggregateTypes(project.AggregateType).
+		AggregateIDs(req.ProjectId).
+		EventData(map[string]interface{}{
+			"grantId": req.GrantId,
+		}).
+		Builder()
+	if asc {
+		query.OrderAsc()
+	}
+
+	changes, err := s.query.SearchEvents(ctx, query, s.auditLogRetention)
 	if err != nil {
 		return nil, err
 	}
+
 	return &mgmt_pb.ListProjectGrantChangesResponse{
-		Result: change_grpc.ChangesToPb(res.Changes, s.assetAPIPrefix(ctx)),
+		Result: change_grpc.EventsToChangesPb(changes, s.assetAPIPrefix(ctx)),
 	}, nil
 }
 
@@ -109,13 +139,38 @@ func (s *Server) ListGrantedProjectRoles(ctx context.Context, req *mgmt_pb.ListG
 }
 
 func (s *Server) ListProjectChanges(ctx context.Context, req *mgmt_pb.ListProjectChangesRequest) (*mgmt_pb.ListProjectChangesResponse, error) {
-	sequence, limit, asc := change_grpc.ChangeQueryToQuery(req.Query)
-	res, err := s.query.ProjectChanges(ctx, req.ProjectId, sequence, limit, asc, s.auditLogRetention)
+	var (
+		limit    uint64
+		sequence uint64
+		asc      bool
+	)
+	if req.Query != nil {
+		limit = uint64(req.Query.Limit)
+		sequence = req.Query.Sequence
+		asc = req.Query.Asc
+	}
+
+	query := eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
+		AllowTimeTravel().
+		Limit(limit).
+		OrderDesc().
+		ResourceOwner(authz.GetCtxData(ctx).OrgID).
+		AddQuery().
+		SequenceGreater(sequence).
+		AggregateTypes(project.AggregateType).
+		AggregateIDs(req.ProjectId).
+		Builder()
+	if asc {
+		query.OrderAsc()
+	}
+
+	changes, err := s.query.SearchEvents(ctx, query, s.auditLogRetention)
 	if err != nil {
 		return nil, err
 	}
+
 	return &mgmt_pb.ListProjectChangesResponse{
-		Result: change_grpc.ChangesToPb(res.Changes, s.assetAPIPrefix(ctx)),
+		Result: change_grpc.EventsToChangesPb(changes, s.assetAPIPrefix(ctx)),
 	}, nil
 }
 

internal/api/grpc/management/project_application.go

@@ -9,7 +9,9 @@ import (
 	object_grpc "github.com/zitadel/zitadel/internal/api/grpc/object"
 	project_grpc "github.com/zitadel/zitadel/internal/api/grpc/project"
 	"github.com/zitadel/zitadel/internal/domain"
+	"github.com/zitadel/zitadel/internal/eventstore"
 	"github.com/zitadel/zitadel/internal/query"
+	"github.com/zitadel/zitadel/internal/repository/project"
 	mgmt_pb "github.com/zitadel/zitadel/pkg/grpc/management"
 )
 
@@ -39,13 +41,41 @@ func (s *Server) ListApps(ctx context.Context, req *mgmt_pb.ListAppsRequest) (*m
 }
 
 func (s *Server) ListAppChanges(ctx context.Context, req *mgmt_pb.ListAppChangesRequest) (*mgmt_pb.ListAppChangesResponse, error) {
-	sequence, limit, asc := change_grpc.ChangeQueryToQuery(req.Query)
-	res, err := s.query.ApplicationChanges(ctx, req.ProjectId, req.AppId, sequence, limit, asc, s.auditLogRetention)
+	var (
+		limit    uint64
+		sequence uint64
+		asc      bool
+	)
+	if req.Query != nil {
+		limit = uint64(req.Query.Limit)
+		sequence = req.Query.Sequence
+		asc = req.Query.Asc
+	}
+
+	query := eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
+		AllowTimeTravel().
+		Limit(limit).
+		OrderDesc().
+		ResourceOwner(authz.GetCtxData(ctx).OrgID).
+		AddQuery().
+		SequenceGreater(sequence).
+		AggregateTypes(project.AggregateType).
+		AggregateIDs(req.ProjectId).
+		EventData(map[string]interface{}{
+			"appId": req.AppId,
+		}).
+		Builder()
+	if asc {
+		query.OrderAsc()
+	}
+
+	changes, err := s.query.SearchEvents(ctx, query, s.auditLogRetention)
 	if err != nil {
 		return nil, err
 	}
+
 	return &mgmt_pb.ListAppChangesResponse{
-		Result: change_grpc.ChangesToPb(res.Changes, s.assetAPIPrefix(ctx)),
+		Result: change_grpc.EventsToChangesPb(changes, s.assetAPIPrefix(ctx)),
 	}, nil
 }
 

internal/api/grpc/management/user.go

@@ -20,7 +20,9 @@ import (
 	"github.com/zitadel/zitadel/internal/api/ui/login"
 	"github.com/zitadel/zitadel/internal/command"
 	"github.com/zitadel/zitadel/internal/domain"
+	"github.com/zitadel/zitadel/internal/eventstore"
 	"github.com/zitadel/zitadel/internal/query"
+	"github.com/zitadel/zitadel/internal/repository/user"
 	mgmt_pb "github.com/zitadel/zitadel/pkg/grpc/management"
 )
 
@@ -73,13 +75,38 @@ func (s *Server) ListUsers(ctx context.Context, req *mgmt_pb.ListUsersRequest) (
 }
 
 func (s *Server) ListUserChanges(ctx context.Context, req *mgmt_pb.ListUserChangesRequest) (*mgmt_pb.ListUserChangesResponse, error) {
-	sequence, limit, asc := change_grpc.ChangeQueryToQuery(req.Query)
-	res, err := s.query.UserChanges(ctx, req.UserId, sequence, limit, asc, s.auditLogRetention)
+	var (
+		limit    uint64
+		sequence uint64
+		asc      bool
+	)
+	if req.Query != nil {
+		limit = uint64(req.Query.Limit)
+		sequence = req.Query.Sequence
+		asc = req.Query.Asc
+	}
+
+	query := eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
+		AllowTimeTravel().
+		Limit(limit).
+		OrderDesc().
+		ResourceOwner(authz.GetCtxData(ctx).OrgID).
+		AddQuery().
+		SequenceGreater(sequence).
+		AggregateTypes(user.AggregateType).
+		AggregateIDs(req.UserId).
+		Builder()
+	if asc {
+		query.OrderAsc()
+	}
+
+	changes, err := s.query.SearchEvents(ctx, query, s.auditLogRetention)
 	if err != nil {
 		return nil, err
 	}
+
 	return &mgmt_pb.ListUserChangesResponse{
-		Result: change_grpc.ChangesToPb(res.Changes, s.assetAPIPrefix(ctx)),
+		Result: change_grpc.EventsToChangesPb(changes, s.assetAPIPrefix(ctx)),
 	}, nil
 }