feat: sequence and timestamp on searchrequests (#468)

* feat: reread events

* feat: sequence and timestamo on search requests

* feat: sequence and timestamo on search requests

* fix: better naming

* fix: log errors

* fix: read sequence before search request
This commit is contained in:
Fabi
2020-07-15 13:24:36 +02:00
committed by GitHub
parent 87155f8c9e
commit 423b86a03b
102 changed files with 7389 additions and 6302 deletions
@@ -2,6 +2,7 @@ package eventstore
import (
"context"
"github.com/caos/logging"
"strings"
"github.com/caos/zitadel/internal/api/authz"
@@ -57,16 +58,23 @@ func (repo *OrgRepository) GetMyOrgIamPolicy(ctx context.Context) (*org_model.Or
func (repo *OrgRepository) SearchMyOrgDomains(ctx context.Context, request *org_model.OrgDomainSearchRequest) (*org_model.OrgDomainSearchResponse, error) {
request.EnsureLimit(repo.SearchLimit)
request.Queries = append(request.Queries, &org_model.OrgDomainSearchQuery{Key: org_model.OrgDomainSearchKeyOrgID, Method: global_model.SearchMethodEquals, Value: authz.GetCtxData(ctx).OrgID})
sequence, err := repo.View.GetLatestOrgDomainSequence()
logging.Log("EVENT-SLowp").OnError(err).Warn("could not read latest org domain sequence")
domains, count, err := repo.View.SearchOrgDomains(request)
if err != nil {
return nil, err
}
return &org_model.OrgDomainSearchResponse{
result := &org_model.OrgDomainSearchResponse{
Offset: request.Offset,
Limit: request.Limit,
TotalResult: uint64(count),
Result: model.OrgDomainsToModel(domains),
}, nil
}
if err == nil {
result.Sequence = sequence.CurrentSequence
result.Timestamp = sequence.CurrentTimestamp
}
return result, nil
}
func (repo *OrgRepository) AddMyOrgDomain(ctx context.Context, domain *org_model.OrgDomain) (*org_model.OrgDomain, error) {
@@ -120,16 +128,23 @@ func (repo *OrgRepository) RemoveMyOrgMember(ctx context.Context, userID string)
func (repo *OrgRepository) SearchMyOrgMembers(ctx context.Context, request *org_model.OrgMemberSearchRequest) (*org_model.OrgMemberSearchResponse, error) {
request.EnsureLimit(repo.SearchLimit)
request.Queries[len(request.Queries)-1] = &org_model.OrgMemberSearchQuery{Key: org_model.OrgMemberSearchKeyOrgID, Method: global_model.SearchMethodEquals, Value: authz.GetCtxData(ctx).OrgID}
sequence, err := repo.View.GetLatestOrgMemberSequence()
logging.Log("EVENT-Smu3d").OnError(err).Warn("could not read latest org member sequence")
members, count, err := repo.View.SearchOrgMembers(request)
if err != nil {
return nil, err
}
return &org_model.OrgMemberSearchResponse{
result := &org_model.OrgMemberSearchResponse{
Offset: request.Offset,
Limit: request.Limit,
TotalResult: uint64(count),
Result: model.OrgMembersToModel(members),
}, nil
}
if err == nil {
result.Sequence = sequence.CurrentSequence
result.Timestamp = sequence.CurrentTimestamp
}
return result, nil
}
func (repo *OrgRepository) GetOrgMemberRoles() []string {
@@ -89,16 +89,23 @@ func (repo *ProjectRepo) SearchProjects(ctx context.Context, request *proj_model
request.Queries = append(request.Queries, &proj_model.ProjectViewSearchQuery{Key: proj_model.ProjectViewSearchKeyProjectID, Method: global_model.SearchMethodIsOneOf, Value: ids})
}
sequence, err := repo.View.GetLatestProjectSequence()
logging.Log("EVENT-Edc56").OnError(err).Warn("could not read latest project sequence")
projects, count, err := repo.View.SearchProjects(request)
if err != nil {
return nil, err
}
return &proj_model.ProjectViewSearchResponse{
result := &proj_model.ProjectViewSearchResponse{
Offset: request.Offset,
Limit: request.Limit,
TotalResult: uint64(count),
Result: model.ProjectsToModel(projects),
}, nil
}
if err == nil {
result.Sequence = sequence.CurrentSequence
result.Timestamp = sequence.CurrentTimestamp
}
return result, nil
}
func (repo *ProjectRepo) ProjectGrantViewByID(ctx context.Context, grantID string) (project *proj_model.ProjectGrantView, err error) {
@@ -132,16 +139,23 @@ func (repo *ProjectRepo) RemoveProjectMember(ctx context.Context, projectID, use
func (repo *ProjectRepo) SearchProjectMembers(ctx context.Context, request *proj_model.ProjectMemberSearchRequest) (*proj_model.ProjectMemberSearchResponse, error) {
request.EnsureLimit(repo.SearchLimit)
sequence, err := repo.View.GetLatestProjectMemberSequence()
logging.Log("EVENT-3dgt6").OnError(err).Warn("could not read latest project member sequence")
members, count, err := repo.View.SearchProjectMembers(request)
if err != nil {
return nil, err
}
return &proj_model.ProjectMemberSearchResponse{
result := &proj_model.ProjectMemberSearchResponse{
Offset: request.Offset,
Limit: request.Limit,
TotalResult: uint64(count),
Result: model.ProjectMembersToModel(members),
}, nil
}
if err == nil {
result.Sequence = sequence.CurrentSequence
result.Timestamp = sequence.CurrentTimestamp
}
return result, nil
}
func (repo *ProjectRepo) AddProjectRole(ctx context.Context, role *proj_model.ProjectRole) (*proj_model.ProjectRole, error) {
@@ -194,18 +208,27 @@ func (repo *ProjectRepo) RemoveProjectRole(ctx context.Context, projectID, key s
return nil
}
func (repo *ProjectRepo) SearchProjectRoles(ctx context.Context, request *proj_model.ProjectRoleSearchRequest) (*proj_model.ProjectRoleSearchResponse, error) {
func (repo *ProjectRepo) SearchProjectRoles(ctx context.Context, projectID string, request *proj_model.ProjectRoleSearchRequest) (*proj_model.ProjectRoleSearchResponse, error) {
request.EnsureLimit(repo.SearchLimit)
request.AppendProjectQuery(projectID)
sequence, err := repo.View.GetLatestProjectRoleSequence()
logging.Log("LSp0d-47suf").OnError(err).Warn("could not read latest project role sequence")
roles, count, err := repo.View.SearchProjectRoles(request)
if err != nil {
return nil, err
}
return &proj_model.ProjectRoleSearchResponse{
result := &proj_model.ProjectRoleSearchResponse{
Offset: request.Offset,
Limit: request.Limit,
TotalResult: uint64(count),
Result: model.ProjectRolesToModel(roles),
}, nil
}
if err == nil {
result.Sequence = sequence.CurrentSequence
result.Timestamp = sequence.CurrentTimestamp
}
return result, nil
}
func (repo *ProjectRepo) ProjectChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool) (*proj_model.ProjectChanges, error) {
@@ -254,16 +277,23 @@ func (repo *ProjectRepo) RemoveApplication(ctx context.Context, projectID, appID
func (repo *ProjectRepo) SearchApplications(ctx context.Context, request *proj_model.ApplicationSearchRequest) (*proj_model.ApplicationSearchResponse, error) {
request.EnsureLimit(repo.SearchLimit)
sequence, err := repo.View.GetLatestApplicationSequence()
logging.Log("EVENT-SKe8s").OnError(err).Warn("could not read latest application sequence")
apps, count, err := repo.View.SearchApplications(request)
if err != nil {
return nil, err
}
return &proj_model.ApplicationSearchResponse{
result := &proj_model.ApplicationSearchResponse{
Offset: request.Offset,
Limit: request.Limit,
TotalResult: uint64(count),
Result: model.ApplicationViewsToModel(apps),
}, nil
}
if err == nil {
result.Sequence = sequence.CurrentSequence
result.Timestamp = sequence.CurrentTimestamp
}
return result, nil
}
func (repo *ProjectRepo) ApplicationChanges(ctx context.Context, id string, appId string, lastSequence uint64, limit uint64, sortAscending bool) (*proj_model.ApplicationChanges, error) {
@@ -299,16 +329,23 @@ func (repo *ProjectRepo) ProjectGrantByID(ctx context.Context, grantID string) (
func (repo *ProjectRepo) SearchProjectGrants(ctx context.Context, request *proj_model.ProjectGrantViewSearchRequest) (*proj_model.ProjectGrantViewSearchResponse, error) {
request.EnsureLimit(repo.SearchLimit)
sequence, err := repo.View.GetLatestProjectGrantSequence()
logging.Log("EVENT-Skw9f").OnError(err).Warn("could not read latest project grant sequence")
projects, count, err := repo.View.SearchProjectGrants(request)
if err != nil {
return nil, err
}
return &proj_model.ProjectGrantViewSearchResponse{
result := &proj_model.ProjectGrantViewSearchResponse{
Offset: request.Offset,
Limit: request.Limit,
TotalResult: uint64(count),
Result: model.ProjectGrantsToModel(projects),
}, nil
}
if err == nil {
result.Sequence = sequence.CurrentSequence
result.Timestamp = sequence.CurrentTimestamp
}
return result, nil
}
func (repo *ProjectRepo) AddProjectGrant(ctx context.Context, grant *proj_model.ProjectGrant) (*proj_model.ProjectGrant, error) {
@@ -430,16 +467,23 @@ func (repo *ProjectRepo) RemoveProjectGrantMember(ctx context.Context, projectID
func (repo *ProjectRepo) SearchProjectGrantMembers(ctx context.Context, request *proj_model.ProjectGrantMemberSearchRequest) (*proj_model.ProjectGrantMemberSearchResponse, error) {
request.EnsureLimit(repo.SearchLimit)
sequence, err := repo.View.GetLatestProjectGrantMemberSequence()
logging.Log("EVENT-Du8sk").OnError(err).Warn("could not read latest project grant sequence")
members, count, err := repo.View.SearchProjectGrantMembers(request)
if err != nil {
return nil, err
}
return &proj_model.ProjectGrantMemberSearchResponse{
result := &proj_model.ProjectGrantMemberSearchResponse{
Offset: request.Offset,
Limit: request.Limit,
TotalResult: uint64(count),
Result: model.ProjectGrantMembersToModel(members),
}, nil
}
if err == nil {
result.Sequence = sequence.CurrentSequence
result.Timestamp = sequence.CurrentTimestamp
}
return result, nil
}
func (repo *ProjectRepo) GetProjectMemberRoles() []string {
@@ -87,16 +87,23 @@ func (repo *UserRepo) UnlockUser(ctx context.Context, id string) (*usr_model.Use
func (repo *UserRepo) SearchUsers(ctx context.Context, request *usr_model.UserSearchRequest) (*usr_model.UserSearchResponse, error) {
request.EnsureLimit(repo.SearchLimit)
sequence, err := repo.View.GetLatestUserSequence()
logging.Log("EVENT-Lcn7d").OnError(err).Warn("could not read latest user sequence")
projects, count, err := repo.View.SearchUsers(request)
if err != nil {
return nil, err
}
return &usr_model.UserSearchResponse{
result := &usr_model.UserSearchResponse{
Offset: request.Offset,
Limit: request.Limit,
TotalResult: uint64(count),
Result: model.UsersToModel(projects),
}, nil
}
if err == nil {
result.Sequence = sequence.CurrentSequence
result.Timestamp = sequence.CurrentTimestamp
}
return result, nil
}
func (repo *UserRepo) UserChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool) (*usr_model.UserChanges, error) {
@@ -2,6 +2,7 @@ package eventstore
import (
"context"
"github.com/caos/logging"
"github.com/caos/zitadel/internal/management/repository/eventsourcing/view"
grant_model "github.com/caos/zitadel/internal/usergrant/model"
grant_event "github.com/caos/zitadel/internal/usergrant/repository/eventsourcing"
@@ -56,14 +57,21 @@ func (repo *UserGrantRepo) BulkRemoveUserGrant(ctx context.Context, grantIDs ...
func (repo *UserGrantRepo) SearchUserGrants(ctx context.Context, request *grant_model.UserGrantSearchRequest) (*grant_model.UserGrantSearchResponse, error) {
request.EnsureLimit(repo.SearchLimit)
sequence, err := repo.View.GetLatestUserGrantSequence()
logging.Log("EVENT-5Viwf").OnError(err).Warn("could not read latest user grant sequence")
grants, count, err := repo.View.SearchUserGrants(request)
if err != nil {
return nil, err
}
return &grant_model.UserGrantSearchResponse{
result := &grant_model.UserGrantSearchResponse{
Offset: request.Offset,
Limit: request.Limit,
TotalResult: uint64(count),
Result: model.UserGrantsToModel(grants),
}, nil
}
if err == nil {
result.Sequence = sequence.CurrentSequence
result.Timestamp = sequence.CurrentTimestamp
}
return result, nil
}
@@ -31,7 +31,7 @@ func (p *Application) EventQuery() (*models.SearchQuery, error) {
if err != nil {
return nil, err
}
return eventsourcing.ProjectQuery(sequence), nil
return eventsourcing.ProjectQuery(sequence.CurrentSequence), nil
}
func (p *Application) Reduce(event *models.Event) (err error) {
@@ -30,7 +30,7 @@ func (o *Org) EventQuery() (*es_models.SearchQuery, error) {
if err != nil {
return nil, err
}
return eventsourcing.OrgQuery(sequence), nil
return eventsourcing.OrgQuery(sequence.CurrentSequence), nil
}
func (o *Org) Reduce(event *es_models.Event) error {
@@ -32,7 +32,7 @@ func (d *OrgDomain) EventQuery() (*models.SearchQuery, error) {
}
return es_models.NewSearchQuery().
AggregateTypeFilter(model.OrgAggregate).
LatestSequenceFilter(sequence), nil
LatestSequenceFilter(sequence.CurrentSequence), nil
}
func (d *OrgDomain) Reduce(event *models.Event) (err error) {
@@ -37,7 +37,7 @@ func (m *OrgMember) EventQuery() (*models.SearchQuery, error) {
}
return es_models.NewSearchQuery().
AggregateTypeFilter(model.OrgAggregate, usr_es_model.UserAggregate).
LatestSequenceFilter(sequence), nil
LatestSequenceFilter(sequence.CurrentSequence), nil
}
func (m *OrgMember) Reduce(event *models.Event) (err error) {
@@ -33,7 +33,7 @@ func (p *Project) EventQuery() (*models.SearchQuery, error) {
if err != nil {
return nil, err
}
return proj_event.ProjectQuery(sequence), nil
return proj_event.ProjectQuery(sequence.CurrentSequence), nil
}
func (p *Project) Reduce(event *models.Event) (err error) {
@@ -39,7 +39,7 @@ func (p *ProjectGrant) EventQuery() (*models.SearchQuery, error) {
if err != nil {
return nil, err
}
return proj_event.ProjectQuery(sequence), nil
return proj_event.ProjectQuery(sequence.CurrentSequence), nil
}
func (p *ProjectGrant) Reduce(event *models.Event) (err error) {
@@ -128,4 +128,3 @@ func (p *ProjectGrant) OnError(event *models.Event, err error) error {
logging.LogWithFields("SPOOL-is8wa", "id", event.AggregateID).WithError(err).Warn("something went wrong in granted projecthandler")
return spooler.HandleError(event, err, p.view.GetLatestProjectGrantFailedEvent, p.view.ProcessedProjectGrantFailedEvent, p.view.ProcessedProjectGrantSequence, p.errorCountUntilSkip)
}
@@ -36,7 +36,7 @@ func (p *ProjectGrantMember) EventQuery() (*models.SearchQuery, error) {
}
return es_models.NewSearchQuery().
AggregateTypeFilter(proj_es_model.ProjectAggregate, usr_es_model.UserAggregate).
LatestSequenceFilter(sequence), nil
LatestSequenceFilter(sequence.CurrentSequence), nil
}
func (p *ProjectGrantMember) Reduce(event *models.Event) (err error) {
@@ -36,7 +36,7 @@ func (p *ProjectMember) EventQuery() (*models.SearchQuery, error) {
}
return es_models.NewSearchQuery().
AggregateTypeFilter(proj_es_model.ProjectAggregate, usr_es_model.UserAggregate).
LatestSequenceFilter(sequence), nil
LatestSequenceFilter(sequence.CurrentSequence), nil
}
func (p *ProjectMember) Reduce(event *models.Event) (err error) {
@@ -31,7 +31,7 @@ func (p *ProjectRole) EventQuery() (*models.SearchQuery, error) {
if err != nil {
return nil, err
}
return eventsourcing.ProjectQuery(sequence), nil
return eventsourcing.ProjectQuery(sequence.CurrentSequence), nil
}
func (p *ProjectRole) Reduce(event *models.Event) (err error) {
@@ -41,7 +41,7 @@ func (p *User) EventQuery() (*models.SearchQuery, error) {
}
return es_models.NewSearchQuery().
AggregateTypeFilter(es_model.UserAggregate, org_es_model.OrgAggregate).
LatestSequenceFilter(sequence), nil
LatestSequenceFilter(sequence.CurrentSequence), nil
}
func (u *User) Reduce(event *models.Event) (err error) {
@@ -48,7 +48,7 @@ func (u *UserGrant) EventQuery() (*models.SearchQuery, error) {
}
return es_models.NewSearchQuery().
AggregateTypeFilter(grant_es_model.UserGrantAggregate, usr_es_model.UserAggregate, proj_es_model.ProjectAggregate).
LatestSequenceFilter(sequence), nil
LatestSequenceFilter(sequence.CurrentSequence), nil
}
func (u *UserGrant) Reduce(event *models.Event) (err error) {
@@ -35,7 +35,7 @@ func (v *View) DeleteApplication(appID string, eventSequence uint64) error {
return v.ProcessedApplicationSequence(eventSequence)
}
func (v *View) GetLatestApplicationSequence() (uint64, error) {
func (v *View) GetLatestApplicationSequence() (*repository.CurrentSequence, error) {
return v.latestSequence(applicationTable)
}
@@ -30,7 +30,7 @@ func (v *View) ProcessedOrgFailedEvent(failedEvent *repository.FailedEvent) erro
return v.saveFailedEvent(failedEvent)
}
func (v *View) GetLatestOrgSequence() (uint64, error) {
func (v *View) GetLatestOrgSequence() (*repository.CurrentSequence, error) {
return v.latestSequence(orgTable)
}
@@ -46,7 +46,7 @@ func (v *View) DeleteOrgDomain(orgID, domain string, eventSequence uint64) error
return v.ProcessedOrgDomainSequence(eventSequence)
}
func (v *View) GetLatestOrgDomainSequence() (uint64, error) {
func (v *View) GetLatestOrgDomainSequence() (*repository.CurrentSequence, error) {
return v.latestSequence(orgDomainTable)
}
@@ -39,7 +39,7 @@ func (v *View) DeleteOrgMember(orgID, userID string, eventSequence uint64) error
return v.ProcessedOrgMemberSequence(eventSequence)
}
func (v *View) GetLatestOrgMemberSequence() (uint64, error) {
func (v *View) GetLatestOrgMemberSequence() (*repository.CurrentSequence, error) {
return v.latestSequence(orgMemberTable)
}
@@ -35,7 +35,7 @@ func (v *View) DeleteProject(projectID string, eventSequence uint64) error {
return v.ProcessedProjectSequence(eventSequence)
}
func (v *View) GetLatestProjectSequence() (uint64, error) {
func (v *View) GetLatestProjectSequence() (*repository.CurrentSequence, error) {
return v.latestSequence(projectTable)
}
@@ -47,7 +47,7 @@ func (v *View) DeleteProjectGrant(grantID string, eventSequence uint64) error {
return v.ProcessedProjectGrantSequence(eventSequence)
}
func (v *View) GetLatestProjectGrantSequence() (uint64, error) {
func (v *View) GetLatestProjectGrantSequence() (*repository.CurrentSequence, error) {
return v.latestSequence(grantedProjectTable)
}
@@ -39,7 +39,7 @@ func (v *View) DeleteProjectGrantMember(grantID, userID string, eventSequence ui
return v.ProcessedProjectGrantMemberSequence(eventSequence)
}
func (v *View) GetLatestProjectGrantMemberSequence() (uint64, error) {
func (v *View) GetLatestProjectGrantMemberSequence() (*repository.CurrentSequence, error) {
return v.latestSequence(projectGrantMemberTable)
}
@@ -39,7 +39,7 @@ func (v *View) DeleteProjectMember(projectID, userID string, eventSequence uint6
return v.ProcessedProjectMemberSequence(eventSequence)
}
func (v *View) GetLatestProjectMemberSequence() (uint64, error) {
func (v *View) GetLatestProjectMemberSequence() (*repository.CurrentSequence, error) {
return v.latestSequence(projectMemberTable)
}
@@ -43,7 +43,7 @@ func (v *View) DeleteProjectRole(projectID, orgID, key string, eventSequence uin
return v.ProcessedProjectRoleSequence(eventSequence)
}
func (v *View) GetLatestProjectRoleSequence() (uint64, error) {
func (v *View) GetLatestProjectRoleSequence() (*repository.CurrentSequence, error) {
return v.latestSequence(projectRoleTable)
}
@@ -12,6 +12,6 @@ func (v *View) saveCurrentSequence(viewName string, sequence uint64) error {
return repository.SaveCurrentSequence(v.Db, sequencesTable, viewName, sequence)
}
func (v *View) latestSequence(viewName string) (uint64, error) {
func (v *View) latestSequence(viewName string) (*repository.CurrentSequence, error) {
return repository.LatestSequence(v.Db, sequencesTable, viewName)
}
@@ -54,7 +54,7 @@ func (v *View) DeleteUser(userID string, eventSequence uint64) error {
return v.ProcessedUserSequence(eventSequence)
}
func (v *View) GetLatestUserSequence() (uint64, error) {
func (v *View) GetLatestUserSequence() (*repository.CurrentSequence, error) {
return v.latestSequence(userTable)
}
@@ -55,7 +55,7 @@ func (v *View) DeleteUserGrant(grantID string, eventSequence uint64) error {
return v.ProcessedUserGrantSequence(eventSequence)
}
func (v *View) GetLatestUserGrantSequence() (uint64, error) {
func (v *View) GetLatestUserGrantSequence() (*repository.CurrentSequence, error) {
return v.latestSequence(userGrantTable)
}
+1 -1
View File
@@ -26,7 +26,7 @@ type ProjectRepository interface {
AddProjectRole(ctx context.Context, role *model.ProjectRole) (*model.ProjectRole, error)
ChangeProjectRole(ctx context.Context, role *model.ProjectRole) (*model.ProjectRole, error)
RemoveProjectRole(ctx context.Context, projectID, key string) error
SearchProjectRoles(ctx context.Context, request *model.ProjectRoleSearchRequest) (*model.ProjectRoleSearchResponse, error)
SearchProjectRoles(ctx context.Context, projectId string, request *model.ProjectRoleSearchRequest) (*model.ProjectRoleSearchResponse, error)
ProjectChanges(ctx context.Context, id string, lastSequence uint64, limit uint64, sortAscending bool) (*model.ProjectChanges, error)
BulkAddProjectRole(ctx context.Context, role []*model.ProjectRole) error