diff --git a/internal/api/call/duration.go b/internal/api/call/duration.go index 24573aa36..b3334de95 100644 --- a/internal/api/call/duration.go +++ b/internal/api/call/duration.go @@ -9,12 +9,18 @@ type durationKey struct{} var key *durationKey = (*durationKey)(nil) -// WithTimestamp sets [time.Now()] adds the call field to the context -// if it's not already set +// WithTimestamp sets [time.Now()] to the call field in the context +// if it's not already set. func WithTimestamp(parent context.Context) context.Context { if parent.Value(key) != nil { return parent } + return ResetTimestamp(parent) +} + +// ResetTimestamp sets [time.Now()] to the call field in the context, +// overwriting any previously set call timestamp. +func ResetTimestamp(parent context.Context) context.Context { return context.WithValue(parent, key, time.Now()) } diff --git a/internal/api/grpc/management/user_integration_test.go b/internal/api/grpc/management/user_integration_test.go new file mode 100644 index 000000000..c1682a5cc --- /dev/null +++ b/internal/api/grpc/management/user_integration_test.go @@ -0,0 +1,87 @@ +//go:build integration + +package management_test + +import ( + "context" + "os" + "testing" + "time" + + "github.com/zitadel/zitadel/internal/integration" + "github.com/zitadel/zitadel/pkg/grpc/management" +) + +var ( + CTX context.Context + Tester *integration.Tester + Client management.ManagementServiceClient +) + +func TestMain(m *testing.M) { + os.Exit(func() int { + ctx, errCtx, cancel := integration.Contexts(3 * time.Minute) + defer cancel() + + Tester = integration.NewTester(ctx) + defer Tester.Done() + + CTX, _ = Tester.WithAuthorization(ctx, integration.OrgOwner), errCtx + Client = Tester.Client.Mgmt + return m.Run() + }()) +} + +// TestImport_and_Get reproduces https://github.com/zitadel/zitadel/issues/5808 +// which led to consistency issues due the call timestamp not being +// updated after a bulk Trigger. +// This test Imports a user and directly tries to Get it, 100 times in a loop. +// When the bug still existed, some (between 1 to 7 out of 100) +// Get calls would return a Not Found error. + +/* Test disabled because it breaks the pipeline. +func TestImport_and_Get(t *testing.T) { + const N = 100 + var misses int + + for i := 0; i < N; i++ { + firstName := strconv.Itoa(i) + t.Run(firstName, func(t *testing.T) { + // create unique names. + lastName := strconv.FormatInt(time.Now().Unix(), 10) + userName := strings.Join([]string{firstName, lastName}, "_") + email := strings.Join([]string{userName, "zitadel.com"}, "@") + + res, err := Client.ImportHumanUser(CTX, &management.ImportHumanUserRequest{ + UserName: userName, + Profile: &management.ImportHumanUserRequest_Profile{ + FirstName: firstName, + LastName: lastName, + PreferredLanguage: language.Afrikaans.String(), + Gender: user.Gender_GENDER_DIVERSE, + }, + Email: &management.ImportHumanUserRequest_Email{ + Email: email, + IsEmailVerified: true, + }, + }) + require.NoError(t, err) + + _, err = Client.GetUserByID(CTX, &management.GetUserByIDRequest{Id: res.GetUserId()}) + + if s, ok := status.FromError(err); ok { + if s == nil { + return + } + if s.Code() == codes.NotFound { + t.Log(s) + misses++ + return + } + } + require.NoError(t, err) // catch and fail on any other error + }) + } + assert.Zerof(t, misses, "Not Found errors %d out of %d", misses, N) +} +*/ diff --git a/internal/api/grpc/session/v2/session.go b/internal/api/grpc/session/v2/session.go index 8ac29a488..d92d873d7 100644 --- a/internal/api/grpc/session/v2/session.go +++ b/internal/api/grpc/session/v2/session.go @@ -16,7 +16,7 @@ import ( ) func (s *Server) GetSession(ctx context.Context, req *session.GetSessionRequest) (*session.GetSessionResponse, error) { - res, err := s.query.SessionByID(ctx, req.GetSessionId(), req.GetSessionToken()) + res, err := s.query.SessionByID(ctx, true, req.GetSessionId(), req.GetSessionToken()) if err != nil { return nil, err } diff --git a/internal/api/grpc/session/v2/session_integration_test.go b/internal/api/grpc/session/v2/session_integration_test.go index 82833d298..fa2918353 100644 --- a/internal/api/grpc/session/v2/session_integration_test.go +++ b/internal/api/grpc/session/v2/session_integration_test.go @@ -10,8 +10,6 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" "github.com/zitadel/zitadel/internal/integration" object "github.com/zitadel/zitadel/pkg/grpc/object/v2alpha" @@ -43,31 +41,16 @@ func TestMain(m *testing.M) { }()) } -func verifyCurrentSession(t testing.TB, id, token string, sequence uint64, window time.Duration, metadata map[string][]byte, factors ...wantFactor) (s *session.Session) { +func verifyCurrentSession(t testing.TB, id, token string, sequence uint64, window time.Duration, metadata map[string][]byte, factors ...wantFactor) *session.Session { require.NotEmpty(t, id) require.NotEmpty(t, token) -retry: - for { - resp, err := Client.GetSession(CTX, &session.GetSessionRequest{ - SessionId: id, - SessionToken: &token, - }) - if err == nil { - s = resp.GetSession() - break retry - } - if code := status.Convert(err).Code(); code == codes.NotFound || code == codes.PermissionDenied { - select { - case <-CTX.Done(): - t.Fatal(CTX.Err(), err) - case <-time.After(time.Second): - t.Log("retrying GetSession") - continue - } - } - require.NoError(t, err) - } + resp, err := Client.GetSession(CTX, &session.GetSessionRequest{ + SessionId: id, + SessionToken: &token, + }) + require.NoError(t, err) + s := resp.GetSession() assert.Equal(t, id, s.GetId()) assert.WithinRange(t, s.GetCreationDate().AsTime(), time.Now().Add(-window), time.Now().Add(window)) diff --git a/internal/eventstore/handler/handler_projection.go b/internal/eventstore/handler/handler_projection.go index c8c309690..baab2d3a6 100644 --- a/internal/eventstore/handler/handler_projection.go +++ b/internal/eventstore/handler/handler_projection.go @@ -6,9 +6,11 @@ import ( "runtime/debug" "time" + "github.com/sirupsen/logrus" "github.com/zitadel/logging" "github.com/zitadel/zitadel/internal/api/authz" + "github.com/zitadel/zitadel/internal/api/call" "github.com/zitadel/zitadel/internal/eventstore" "github.com/zitadel/zitadel/internal/repository/pseudo" ) @@ -110,27 +112,51 @@ func NewProjectionHandler( return h } -// Trigger handles all events for the provided instances (or current instance from context if non specified) -// by calling FetchEvents and Process until the amount of events is smaller than the BulkLimit -func (h *ProjectionHandler) Trigger(ctx context.Context, instances ...string) error { - ids := []string{authz.GetInstance(ctx).InstanceID()} - if len(instances) > 0 { - ids = instances +func triggerInstances(ctx context.Context, instances []string) []string { + if len(instances) == 0 { + instances = append(instances, authz.GetInstance(ctx).InstanceID()) } + return instances +} + +// Trigger handles all events for the provided instances (or current instance from context if non specified) +// by calling FetchEvents and Process until the amount of events is smaller than the BulkLimit. +// If a bulk action was executed, the call timestamp in context will be reset for subsequent queries. +// The returned context is never nil. It is either the original context or an updated context. +// +// If Trigger encounters an error, it is only logged. If the error is important for the caller, +// use TriggerErr instead. +func (h *ProjectionHandler) Trigger(ctx context.Context, instances ...string) context.Context { + instances = triggerInstances(ctx, instances) + ctx, err := h.TriggerErr(ctx, instances...) + logging.OnError(err).WithFields(logrus.Fields{ + "projection": h.ProjectionName, + "instanceIDs": instances, + }).Error("trigger failed") + return ctx +} + +// TriggerErr handles all events for the provided instances (or current instance from context if non specified) +// by calling FetchEvents and Process until the amount of events is smaller than the BulkLimit. +// If a bulk action was executed, the call timestamp in context will be reset for subsequent queries. +// The returned context is never nil. It is either the original context or an updated context. +func (h *ProjectionHandler) TriggerErr(ctx context.Context, instances ...string) (context.Context, error) { + instances = triggerInstances(ctx, instances) for { - events, hasLimitExceeded, err := h.FetchEvents(ctx, ids...) + events, hasLimitExceeded, err := h.FetchEvents(ctx, instances...) if err != nil { - return err + return ctx, err } if len(events) == 0 { - return nil + return ctx, nil } _, err = h.Process(ctx, events...) + ctx = call.ResetTimestamp(ctx) if err != nil { - return err + return ctx, err } if !hasLimitExceeded { - return nil + return ctx, nil } } } @@ -274,7 +300,7 @@ func (h *ProjectionHandler) schedule(ctx context.Context) { continue } go h.cancelOnErr(lockInstanceCtx, errs, cancelInstanceLock) - err = h.Trigger(lockInstanceCtx, instances...) + _, err = h.TriggerErr(lockInstanceCtx, instances...) if err != nil { logging.WithFields("projection", h.ProjectionName, "instanceIDs", instances).WithError(err).Error("trigger failed") failed = true diff --git a/internal/eventstore/handler/handler_projection_test.go b/internal/eventstore/handler/handler_projection_test.go index 349799e46..3ee791a2d 100644 --- a/internal/eventstore/handler/handler_projection_test.go +++ b/internal/eventstore/handler/handler_projection_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/zitadel/zitadel/internal/api/authz" + "github.com/zitadel/zitadel/internal/api/call" "github.com/zitadel/zitadel/internal/api/service" "github.com/zitadel/zitadel/internal/database" "github.com/zitadel/zitadel/internal/eventstore" @@ -29,6 +30,47 @@ var ( ) func TestProjectionHandler_Trigger(t *testing.T) { + const pause = time.Millisecond + + startCtx := call.WithTimestamp(context.Background()) + start := call.FromContext(startCtx) + + h := &ProjectionHandler{ + Handler: Handler{ + Eventstore: eventstore.NewEventstore(eventstore.TestConfig( + es_repo_mock.NewRepo(t).ExpectFilterEvents( + &repository.Event{ + ID: "id", + Sequence: 1, + PreviousAggregateSequence: 0, + CreationDate: time.Now(), + Type: "test.added", + Version: "v1", + AggregateID: "testid", + AggregateType: "testAgg", + }, + ), + )), + }, + ProjectionName: "test", + reduce: testReduce(newTestStatement("testAgg", 1, 0)), + update: testUpdate(t, 1, 0, nil), + searchQuery: testQuery( + eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent). + AddQuery(). + AggregateTypes("test"). + Builder(), + 5, nil, + ), + } + + time.Sleep(pause) + endCtx := h.Trigger(startCtx) + // check if the new context has a call timestamp that's later than start+pause. + assert.WithinRange(t, call.FromContext(endCtx), start.Add(pause), start.Add(pause+time.Second)) +} + +func TestProjectionHandler_TriggerErr(t *testing.T) { type fields struct { reduce Reduce update Update @@ -223,7 +265,8 @@ func TestProjectionHandler_Trigger(t *testing.T) { searchQuery: tt.fields.query, } - err := h.Trigger(tt.args.ctx, tt.args.instances...) + // context timestamp is checked in [TestProjectionHandler_Trigger] + _, err := h.TriggerErr(tt.args.ctx, tt.args.instances...) if !tt.want.isErr(err) { t.Errorf("unexpected error %v", err) } diff --git a/internal/query/app.go b/internal/query/app.go index 6b40f712e..577f6c181 100644 --- a/internal/query/app.go +++ b/internal/query/app.go @@ -253,7 +253,7 @@ func (q *Queries) AppByProjectAndAppID(ctx context.Context, shouldTriggerBulk bo defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.AppProjection.Trigger(ctx) + ctx = projection.AppProjection.Trigger(ctx) } stmt, scan := prepareAppQuery(ctx, q.client) diff --git a/internal/query/authn_key.go b/internal/query/authn_key.go index 3d9341c94..87140b216 100644 --- a/internal/query/authn_key.go +++ b/internal/query/authn_key.go @@ -191,7 +191,7 @@ func (q *Queries) GetAuthNKeyByID(ctx context.Context, shouldTriggerBulk bool, i defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.AuthNKeyProjection.Trigger(ctx) + ctx = projection.AuthNKeyProjection.Trigger(ctx) } query, scan := prepareAuthNKeyQuery(ctx, q.client) diff --git a/internal/query/domain_policy.go b/internal/query/domain_policy.go index f9f5f94b7..2f7a596cd 100644 --- a/internal/query/domain_policy.go +++ b/internal/query/domain_policy.go @@ -91,7 +91,7 @@ func (q *Queries) DomainPolicyByOrg(ctx context.Context, shouldTriggerBulk bool, defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.DomainPolicyProjection.Trigger(ctx) + ctx = projection.DomainPolicyProjection.Trigger(ctx) } eq := sq.And{ sq.Eq{DomainPolicyColInstanceID.identifier(): authz.GetInstance(ctx).InstanceID()}, diff --git a/internal/query/idp.go b/internal/query/idp.go index e483755c8..35c0b7b5d 100644 --- a/internal/query/idp.go +++ b/internal/query/idp.go @@ -193,7 +193,7 @@ func (q *Queries) IDPByIDAndResourceOwner(ctx context.Context, shouldTriggerBulk defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.IDPProjection.Trigger(ctx) + ctx = projection.IDPProjection.Trigger(ctx) } eq := sq.Eq{ diff --git a/internal/query/idp_template.go b/internal/query/idp_template.go index 2639691a7..b83efe156 100644 --- a/internal/query/idp_template.go +++ b/internal/query/idp_template.go @@ -7,7 +7,6 @@ import ( "time" sq "github.com/Masterminds/squirrel" - "github.com/zitadel/logging" "github.com/zitadel/zitadel/internal/api/authz" "github.com/zitadel/zitadel/internal/api/call" @@ -612,8 +611,7 @@ func (q *Queries) IDPTemplateByID(ctx context.Context, shouldTriggerBulk bool, i defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - err := projection.IDPTemplateProjection.Trigger(ctx) - logging.OnError(err).WithField("projection", idpTemplateTable.identifier()).Warn("could not trigger projection for query") + ctx = projection.IDPTemplateProjection.Trigger(ctx) } eq := sq.Eq{ diff --git a/internal/query/instance.go b/internal/query/instance.go index 2b4cfb185..9040478af 100644 --- a/internal/query/instance.go +++ b/internal/query/instance.go @@ -182,7 +182,7 @@ func (q *Queries) Instance(ctx context.Context, shouldTriggerBulk bool) (_ *Inst defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.InstanceProjection.Trigger(ctx) + ctx = projection.InstanceProjection.Trigger(ctx) } stmt, scan := prepareInstanceDomainQuery(ctx, q.client, authz.GetInstance(ctx).RequestedDomain()) diff --git a/internal/query/lockout_policy.go b/internal/query/lockout_policy.go index 84b0e27ee..ff9dfc663 100644 --- a/internal/query/lockout_policy.go +++ b/internal/query/lockout_policy.go @@ -86,7 +86,7 @@ func (q *Queries) LockoutPolicyByOrg(ctx context.Context, shouldTriggerBulk bool defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.LockoutPolicyProjection.Trigger(ctx) + ctx = projection.LockoutPolicyProjection.Trigger(ctx) } eq := sq.Eq{ LockoutColInstanceID.identifier(): authz.GetInstance(ctx).InstanceID(), diff --git a/internal/query/login_policy.go b/internal/query/login_policy.go index d575a26d3..4862aa765 100644 --- a/internal/query/login_policy.go +++ b/internal/query/login_policy.go @@ -166,7 +166,7 @@ func (q *Queries) LoginPolicyByID(ctx context.Context, shouldTriggerBulk bool, o defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.LoginPolicyProjection.Trigger(ctx) + ctx = projection.LoginPolicyProjection.Trigger(ctx) } eq := sq.Eq{LoginPolicyColumnInstanceID.identifier(): authz.GetInstance(ctx).InstanceID()} if !withOwnerRemoved { diff --git a/internal/query/notification_policy.go b/internal/query/notification_policy.go index e15e7c84f..64873e1ca 100644 --- a/internal/query/notification_policy.go +++ b/internal/query/notification_policy.go @@ -81,7 +81,8 @@ func (q *Queries) NotificationPolicyByOrg(ctx context.Context, shouldTriggerBulk defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - if err := projection.NotificationPolicyProjection.Trigger(ctx); err != nil { + ctx, err = projection.NotificationPolicyProjection.TriggerErr(ctx) + if err != nil { return nil, err } } @@ -112,7 +113,8 @@ func (q *Queries) DefaultNotificationPolicy(ctx context.Context, shouldTriggerBu defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - if err := projection.NotificationPolicyProjection.Trigger(ctx); err != nil { + ctx, err = projection.NotificationPolicyProjection.TriggerErr(ctx) + if err != nil { return nil, err } } diff --git a/internal/query/org.go b/internal/query/org.go index d5c90f08f..40bcbe400 100644 --- a/internal/query/org.go +++ b/internal/query/org.go @@ -94,7 +94,7 @@ func (q *Queries) OrgByID(ctx context.Context, shouldTriggerBulk bool, id string defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.OrgProjection.Trigger(ctx) + ctx = projection.OrgProjection.Trigger(ctx) } stmt, scan := prepareOrgQuery(ctx, q.client) diff --git a/internal/query/org_metadata.go b/internal/query/org_metadata.go index c22bd6a71..5ce36369e 100644 --- a/internal/query/org_metadata.go +++ b/internal/query/org_metadata.go @@ -82,7 +82,7 @@ func (q *Queries) GetOrgMetadataByKey(ctx context.Context, shouldTriggerBulk boo defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.OrgMetadataProjection.Trigger(ctx) + ctx = projection.OrgMetadataProjection.Trigger(ctx) } query, scan := prepareOrgMetadataQuery(ctx, q.client) @@ -111,7 +111,7 @@ func (q *Queries) SearchOrgMetadata(ctx context.Context, shouldTriggerBulk bool, defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.OrgMetadataProjection.Trigger(ctx) + ctx = projection.OrgMetadataProjection.Trigger(ctx) } eq := sq.Eq{ OrgMetadataOrgIDCol.identifier(): orgID, diff --git a/internal/query/password_age_policy.go b/internal/query/password_age_policy.go index 2de71235a..fda59e88e 100644 --- a/internal/query/password_age_policy.go +++ b/internal/query/password_age_policy.go @@ -86,7 +86,7 @@ func (q *Queries) PasswordAgePolicyByOrg(ctx context.Context, shouldTriggerBulk defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.PasswordAgeProjection.Trigger(ctx) + ctx = projection.PasswordAgeProjection.Trigger(ctx) } eq := sq.Eq{PasswordAgeColInstanceID.identifier(): authz.GetInstance(ctx).InstanceID()} if !withOwnerRemoved { @@ -116,7 +116,7 @@ func (q *Queries) DefaultPasswordAgePolicy(ctx context.Context, shouldTriggerBul defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.PasswordAgeProjection.Trigger(ctx) + ctx = projection.PasswordAgeProjection.Trigger(ctx) } stmt, scan := preparePasswordAgePolicyQuery(ctx, q.client) diff --git a/internal/query/password_complexity_policy.go b/internal/query/password_complexity_policy.go index 4877f0159..e5f15f181 100644 --- a/internal/query/password_complexity_policy.go +++ b/internal/query/password_complexity_policy.go @@ -38,7 +38,7 @@ func (q *Queries) PasswordComplexityPolicyByOrg(ctx context.Context, shouldTrigg defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.PasswordComplexityProjection.Trigger(ctx) + ctx = projection.PasswordComplexityProjection.Trigger(ctx) } eq := sq.Eq{PasswordComplexityColInstanceID.identifier(): authz.GetInstance(ctx).InstanceID()} if !withOwnerRemoved { @@ -68,7 +68,7 @@ func (q *Queries) DefaultPasswordComplexityPolicy(ctx context.Context, shouldTri defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.PasswordComplexityProjection.Trigger(ctx) + ctx = projection.PasswordComplexityProjection.Trigger(ctx) } stmt, scan := preparePasswordComplexityPolicyQuery(ctx, q.client) diff --git a/internal/query/privacy_policy.go b/internal/query/privacy_policy.go index 2874c6d22..e207c3e9b 100644 --- a/internal/query/privacy_policy.go +++ b/internal/query/privacy_policy.go @@ -96,7 +96,7 @@ func (q *Queries) PrivacyPolicyByOrg(ctx context.Context, shouldTriggerBulk bool defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.PrivacyPolicyProjection.Trigger(ctx) + ctx = projection.PrivacyPolicyProjection.Trigger(ctx) } eq := sq.Eq{PrivacyColInstanceID.identifier(): authz.GetInstance(ctx).InstanceID()} if !withOwnerRemoved { @@ -125,7 +125,7 @@ func (q *Queries) DefaultPrivacyPolicy(ctx context.Context, shouldTriggerBulk bo defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.PrivacyPolicyProjection.Trigger(ctx) + ctx = projection.PrivacyPolicyProjection.Trigger(ctx) } stmt, scan := preparePrivacyPolicyQuery(ctx, q.client) diff --git a/internal/query/project.go b/internal/query/project.go index e7df58cf5..065a3e5c5 100644 --- a/internal/query/project.go +++ b/internal/query/project.go @@ -105,7 +105,7 @@ func (q *Queries) ProjectByID(ctx context.Context, shouldTriggerBulk bool, id st defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.ProjectProjection.Trigger(ctx) + ctx = projection.ProjectProjection.Trigger(ctx) } stmt, scan := prepareProjectQuery(ctx, q.client) diff --git a/internal/query/project_grant.go b/internal/query/project_grant.go index bbf265736..fee8ae836 100644 --- a/internal/query/project_grant.go +++ b/internal/query/project_grant.go @@ -116,7 +116,7 @@ func (q *Queries) ProjectGrantByID(ctx context.Context, shouldTriggerBulk bool, defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.ProjectGrantProjection.Trigger(ctx) + ctx = projection.ProjectGrantProjection.Trigger(ctx) } stmt, scan := prepareProjectGrantQuery(ctx, q.client) diff --git a/internal/query/project_role.go b/internal/query/project_role.go index 03ebbf60b..7365b1247 100644 --- a/internal/query/project_role.go +++ b/internal/query/project_role.go @@ -88,7 +88,7 @@ func (q *Queries) SearchProjectRoles(ctx context.Context, shouldTriggerBulk bool defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.ProjectRoleProjection.Trigger(ctx) + ctx = projection.ProjectRoleProjection.Trigger(ctx) } eq := sq.Eq{ProjectRoleColumnInstanceID.identifier(): authz.GetInstance(ctx).InstanceID()} diff --git a/internal/query/session.go b/internal/query/session.go index e9706410a..76ccefba6 100644 --- a/internal/query/session.go +++ b/internal/query/session.go @@ -141,10 +141,14 @@ var ( } ) -func (q *Queries) SessionByID(ctx context.Context, id, sessionToken string) (_ *Session, err error) { +func (q *Queries) SessionByID(ctx context.Context, shouldTriggerBulk bool, id, sessionToken string) (_ *Session, err error) { ctx, span := tracing.NewSpan(ctx) defer func() { span.EndWithError(err) }() + if shouldTriggerBulk { + ctx = projection.SessionProjection.Trigger(ctx) + } + query, scan := prepareSessionQuery(ctx, q.client) stmt, args, err := query.Where( sq.Eq{ diff --git a/internal/query/user.go b/internal/query/user.go index 1550a4678..46a683e67 100644 --- a/internal/query/user.go +++ b/internal/query/user.go @@ -338,8 +338,8 @@ func (q *Queries) GetUserByID(ctx context.Context, shouldTriggerBulk bool, userI defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.UserProjection.Trigger(ctx) - projection.LoginNameProjection.Trigger(ctx) + ctx = projection.UserProjection.Trigger(ctx) + ctx = projection.LoginNameProjection.Trigger(ctx) } query, scan := prepareUserQuery(ctx, q.client) @@ -367,8 +367,8 @@ func (q *Queries) GetUser(ctx context.Context, shouldTriggerBulk bool, withOwner defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.UserProjection.Trigger(ctx) - projection.LoginNameProjection.Trigger(ctx) + ctx = projection.UserProjection.Trigger(ctx) + ctx = projection.LoginNameProjection.Trigger(ctx) } query, scan := prepareUserQuery(ctx, q.client) @@ -467,8 +467,8 @@ func (q *Queries) GetNotifyUserByID(ctx context.Context, shouldTriggered bool, u defer func() { span.EndWithError(err) }() if shouldTriggered { - projection.UserProjection.Trigger(ctx) - projection.LoginNameProjection.Trigger(ctx) + ctx = projection.UserProjection.Trigger(ctx) + ctx = projection.LoginNameProjection.Trigger(ctx) } query, scan := prepareNotifyUserQuery(ctx, q.client) @@ -496,8 +496,8 @@ func (q *Queries) GetNotifyUser(ctx context.Context, shouldTriggered bool, withO defer func() { span.EndWithError(err) }() if shouldTriggered { - projection.UserProjection.Trigger(ctx) - projection.LoginNameProjection.Trigger(ctx) + ctx = projection.UserProjection.Trigger(ctx) + ctx = projection.LoginNameProjection.Trigger(ctx) } query, scan := prepareNotifyUserQuery(ctx, q.client) diff --git a/internal/query/user_grant.go b/internal/query/user_grant.go index ceb8652e4..6d0eaa95b 100644 --- a/internal/query/user_grant.go +++ b/internal/query/user_grant.go @@ -8,8 +8,6 @@ import ( sq "github.com/Masterminds/squirrel" - "github.com/zitadel/logging" - "github.com/zitadel/zitadel/internal/api/authz" "github.com/zitadel/zitadel/internal/api/call" "github.com/zitadel/zitadel/internal/database" @@ -235,7 +233,7 @@ func (q *Queries) UserGrant(ctx context.Context, shouldTriggerBulk bool, withOwn defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.UserGrantProjection.Trigger(ctx) + ctx = projection.UserGrantProjection.Trigger(ctx) } query, scan := prepareUserGrantQuery(ctx, q.client) @@ -260,9 +258,7 @@ func (q *Queries) UserGrants(ctx context.Context, queries *UserGrantsQueries, sh defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - logging.OnError( - projection.UserGrantProjection.Trigger(ctx), - ).Debug("unable to trigger") + ctx = projection.UserGrantProjection.Trigger(ctx) } query, scan := prepareUserGrantsQuery(ctx, q.client) diff --git a/internal/query/user_metadata.go b/internal/query/user_metadata.go index b7c4c2736..d0a9202b8 100644 --- a/internal/query/user_metadata.go +++ b/internal/query/user_metadata.go @@ -82,7 +82,7 @@ func (q *Queries) GetUserMetadataByKey(ctx context.Context, shouldTriggerBulk bo defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.UserMetadataProjection.Trigger(ctx) + ctx = projection.UserMetadataProjection.Trigger(ctx) } query, scan := prepareUserMetadataQuery(ctx, q.client) @@ -111,7 +111,7 @@ func (q *Queries) SearchUserMetadata(ctx context.Context, shouldTriggerBulk bool defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.UserMetadataProjection.Trigger(ctx) + ctx = projection.UserMetadataProjection.Trigger(ctx) } query, scan := prepareUserMetadataListQuery(ctx, q.client) diff --git a/internal/query/user_personal_access_token.go b/internal/query/user_personal_access_token.go index 4afab7ec8..5558eed3c 100644 --- a/internal/query/user_personal_access_token.go +++ b/internal/query/user_personal_access_token.go @@ -90,7 +90,7 @@ func (q *Queries) PersonalAccessTokenByID(ctx context.Context, shouldTriggerBulk defer func() { span.EndWithError(err) }() if shouldTriggerBulk { - projection.PersonalAccessTokenProjection.Trigger(ctx) + ctx = projection.PersonalAccessTokenProjection.Trigger(ctx) } query, scan := preparePersonalAccessTokenQuery(ctx, q.client)