Skip to content

feat(notifier): sending notification concurrently #136

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 9 commits into from
Jun 3, 2024
Merged

feat(notifier): sending notification concurrently #136

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
d05bbc7
feat: sending notification concurrently
Mar 18, 2024
b059882
test: update unit test
Mar 18, 2024
630aec0
test: fix unit test
Mar 20, 2024
fa2eb3c
test: fix race data
Mar 20, 2024
40e7cf6
chore: update ctx
Apr 25, 2024
ff37298
test: update test
Apr 26, 2024
21554ec
test: adding sleep for temporary solution
Apr 26, 2024
e35bce9
Merge remote-tracking branch 'origin/main' into sending-notif-async
Jun 3, 2024
90024a4
test: update test sleep
Jun 3, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
81 changes: 45 additions & 36 deletions core/appeal/service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -408,11 +408,14 @@ func (s *Service) Create(ctx context.Context, appeals []*domain.Appeal, opts ...
}

if len(notifications) > 0 {
if errs := s.notifier.Notify(ctx, notifications); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error())
go func() {
ctx := context.WithoutCancel(ctx)
if errs := s.notifier.Notify(ctx, notifications); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error())
}
}
}
}()
}

return nil
Expand Down Expand Up @@ -622,11 +625,14 @@ func (s *Service) UpdateApproval(ctx context.Context, approvalAction domain.Appr
notifications = append(notifications, s.getApprovalNotifications(ctx, appeal)...)
}
if len(notifications) > 0 {
if errs := s.notifier.Notify(ctx, notifications); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error())
go func() {
ctx := context.WithoutCancel(ctx)
if errs := s.notifier.Notify(ctx, notifications); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error())
}
}
}
}()
}

var auditKey string
Expand Down Expand Up @@ -734,37 +740,40 @@ func (s *Service) AddApprover(ctx context.Context, appealID, approvalID, email s
}
}

if errs := s.notifier.Notify(ctx, []domain.Notification{
{
User: email,
Labels: map[string]string{
"appeal_id": appeal.ID,
},
Message: domain.NotificationMessage{
Type: domain.NotificationTypeApproverNotification,
Variables: map[string]interface{}{
"resource_name": fmt.Sprintf("%s (%s: %s)", appeal.Resource.Name, appeal.Resource.ProviderType, appeal.Resource.URN),
"role": appeal.Role,
"requestor": appeal.CreatedBy,
"appeal_id": appeal.ID,
"account_id": appeal.AccountID,
"account_type": appeal.AccountType,
"provider_type": appeal.Resource.ProviderType,
"resource_type": appeal.Resource.Type,
"created_at": appeal.CreatedAt,
"approval_step": approval.Name,
"actor": email,
"details": appeal.Details,
"duration": duration,
"creator": appeal.Creator,
go func() {
ctx := context.WithoutCancel(ctx)
if errs := s.notifier.Notify(ctx, []domain.Notification{
{
User: email,
Labels: map[string]string{
"appeal_id": appeal.ID,
},
Message: domain.NotificationMessage{
Type: domain.NotificationTypeApproverNotification,
Variables: map[string]interface{}{
"resource_name": fmt.Sprintf("%s (%s: %s)", appeal.Resource.Name, appeal.Resource.ProviderType, appeal.Resource.URN),
"role": appeal.Role,
"requestor": appeal.CreatedBy,
"appeal_id": appeal.ID,
"account_id": appeal.AccountID,
"account_type": appeal.AccountType,
"provider_type": appeal.Resource.ProviderType,
"resource_type": appeal.Resource.Type,
"created_at": appeal.CreatedAt,
"approval_step": approval.Name,
"actor": email,
"details": appeal.Details,
"duration": duration,
"creator": appeal.Creator,
},
},
},
},
}); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error())
}); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error())
}
}
}
}()

return appeal, nil
}
Expand Down
20 changes: 13 additions & 7 deletions core/appeal/service_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1096,6 +1096,7 @@ func (s *ServiceTestSuite) TestCreate() {
Log(mock.Anything, appeal.AuditKeyBulkInsert, mock.Anything).Return(nil).Once()

actualError := h.service.Create(context.Background(), appeals)
time.Sleep(time.Millisecond)

s.Nil(actualError)
s.Equal(expectedResult, appeals)
Expand Down Expand Up @@ -1560,6 +1561,7 @@ func (s *ServiceTestSuite) TestCreate() {
Return(nil).Once()

actualError := h.service.Create(context.Background(), appeals)
time.Sleep(time.Millisecond)

s.Nil(actualError)
s.Equal(expectedResult, appeals)
Expand Down Expand Up @@ -1657,6 +1659,7 @@ func (s *ServiceTestSuite) TestCreate() {
h.mockProviderService.EXPECT().GrantAccess(mock.Anything, mock.Anything).Return(nil).Once()

err := h.service.Create(context.Background(), []*domain.Appeal{input}, appeal.CreateWithAdditionalAppeal())
time.Sleep(time.Millisecond)

s.NoError(err)
s.Equal("test-approval", input.Approvals[0].Name)
Expand Down Expand Up @@ -1933,6 +1936,7 @@ func (s *ServiceTestSuite) TestCreateAppeal__WithExistingAppealAndWithAutoApprov
h.mockAuditLogger.EXPECT().Log(mock.Anything, appeal.AuditKeyBulkInsert, mock.Anything).Return(nil).Once()

actualError := h.service.Create(context.Background(), appeals)
time.Sleep(time.Millisecond)

s.Nil(actualError)
s.Equal(expectedResult, appeals)
Expand Down Expand Up @@ -2129,6 +2133,7 @@ func (s *ServiceTestSuite) TestCreateAppeal__WithAdditionalAppeals() {
h.mockNotifier.EXPECT().Notify(h.ctxMatcher, mock.Anything).Return(nil).Once()

err := h.service.Create(context.Background(), appealsPayload)
time.Sleep(time.Millisecond)

s.NoError(err)
}
Expand Down Expand Up @@ -2763,6 +2768,7 @@ func (s *ServiceTestSuite) TestUpdateApproval() {
Return(nil).Once()

_, actualError := h.service.UpdateApproval(context.Background(), action)
time.Sleep(time.Millisecond)

s.Nil(actualError)
})
Expand Down Expand Up @@ -3164,7 +3170,7 @@ func (s *ServiceTestSuite) TestUpdateApproval() {
Return(nil).Once()

actualResult, actualError := h.service.UpdateApproval(context.Background(), tc.expectedApprovalAction)

time.Sleep(time.Millisecond)
s.NoError(actualError)
tc.expectedResult.Policy = actualResult.Policy
s.Equal(tc.expectedResult, actualResult)
Expand Down Expand Up @@ -3298,8 +3304,6 @@ func (s *ServiceTestSuite) TestCancel() {

func (s *ServiceTestSuite) TestAddApprover() {
s.Run("should return appeal on success", func() {
h := newServiceTestHelper()
defer h.assertExpectations(s.T())
appealID := uuid.New().String()
approvalID := uuid.New().String()
approvalName := "test-approval-name"
Expand All @@ -3320,6 +3324,8 @@ func (s *ServiceTestSuite) TestAddApprover() {

for _, tc := range testCases {
s.Run(tc.name, func() {
h := newServiceTestHelper()
defer h.assertExpectations(s.T())
expectedAppeal := &domain.Appeal{
ID: appealID,
Status: domain.AppealStatusPending,
Expand Down Expand Up @@ -3355,18 +3361,18 @@ func (s *ServiceTestSuite) TestAddApprover() {
h.mockNotifier.EXPECT().
Notify(h.ctxMatcher, mock.Anything).
Run(func(ctx context.Context, notifications []domain.Notification) {
s.Len(notifications, 1)
assert.Equal(s.T(), len(notifications), 1)
n := notifications[0]
s.Equal(tc.newApprover, n.User)
s.Equal(domain.NotificationTypeApproverNotification, n.Message.Type)
assert.Equal(s.T(), tc.newApprover, n.User)
assert.Equal(s.T(), domain.NotificationTypeApproverNotification, n.Message.Type)
}).
Return(nil).Once()

actualAppeal, actualError := h.service.AddApprover(context.Background(), appealID, approvalID, newApprover)

time.Sleep(time.Millisecond)
s.NoError(actualError)
s.Equal(expectedApproval, actualAppeal.Approvals[0])

})
}
})
Expand Down
136 changes: 73 additions & 63 deletions core/grant/service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,37 +140,40 @@ func (s *Service) Update(ctx context.Context, payload *domain.Grant) error {
}

if previousOwner != updatedGrant.Owner {
message := domain.NotificationMessage{
Type: domain.NotificationTypeGrantOwnerChanged,
Variables: map[string]interface{}{
"grant_id": grantDetails.ID,
"previous_owner": previousOwner,
"new_owner": updatedGrant.Owner,
},
}
notifications := []domain.Notification{{
User: updatedGrant.Owner,
Labels: map[string]string{
"appeal_id": grantDetails.AppealID,
"grant_id": grantDetails.ID,
},
Message: message,
}}
if previousOwner != "" {
notifications = append(notifications, domain.Notification{
User: previousOwner,
go func() {
message := domain.NotificationMessage{
Type: domain.NotificationTypeGrantOwnerChanged,
Variables: map[string]interface{}{
"grant_id": grantDetails.ID,
"previous_owner": previousOwner,
"new_owner": updatedGrant.Owner,
},
}
notifications := []domain.Notification{{
User: updatedGrant.Owner,
Labels: map[string]string{
"appeal_id": grantDetails.AppealID,
"grant_id": grantDetails.ID,
},
Message: message,
})
}
if errs := s.notifier.Notify(ctx, notifications); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error())
}}
if previousOwner != "" {
notifications = append(notifications, domain.Notification{
User: previousOwner,
Labels: map[string]string{
"appeal_id": grantDetails.AppealID,
"grant_id": grantDetails.ID,
},
Message: message,
})
}
}
ctx := context.WithoutCancel(ctx)
if errs := s.notifier.Notify(ctx, notifications); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error())
}
}
}()
}

return nil
Expand Down Expand Up @@ -218,7 +221,7 @@ func (s *Service) Revoke(ctx context.Context, id, actor, reason string, opts ...
}

if !options.skipNotification {
if errs := s.notifier.Notify(ctx, []domain.Notification{{
notifications := []domain.Notification{{
User: grant.CreatedBy,
Labels: map[string]string{
"appeal_id": grant.AppealID,
Expand All @@ -235,11 +238,15 @@ func (s *Service) Revoke(ctx context.Context, id, actor, reason string, opts ...
"revoke_reason": grant.RevokeReason,
},
},
}}); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error())
}}
go func() {
ctx := context.WithoutCancel(ctx)
if errs := s.notifier.Notify(ctx, notifications); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error())
}
}
}
}()
}

s.logger.Info(ctx, "grant revoked", "grant_id", id)
Expand Down Expand Up @@ -571,44 +578,47 @@ func (s *Service) DormancyCheck(ctx context.Context, criteria domain.DormancyChe
return fmt.Errorf("updating grants expiration date: %w", err)
}

var notifications []domain.Notification
prepare_notifications:
for owner, grants := range dormantGrantsByOwner {
var grantsMap []map[string]interface{}
var grantIDs []string

for _, g := range grants {
grantMap, err := utils.StructToMap(g)
if err != nil {
s.logger.Error(ctx, "failed to convert grant to map", "error", err)
continue prepare_notifications
go func() {
ctx := context.WithoutCancel(ctx)
var notifications []domain.Notification
prepare_notifications:
for owner, grants := range dormantGrantsByOwner {
var grantsMap []map[string]interface{}
var grantIDs []string

for _, g := range grants {
grantMap, err := utils.StructToMap(g)
if err != nil {
s.logger.Error(ctx, "failed to convert grant to map", "error", err)
continue prepare_notifications
}
grantsMap = append(grantsMap, grantMap)
}
grantsMap = append(grantsMap, grantMap)
}

notifications = append(notifications, domain.Notification{
User: owner,
Labels: map[string]string{
"owner": owner,
"grant_ids": strings.Join(grantIDs, ", "),
},
Message: domain.NotificationMessage{
Type: domain.NotificationTypeUnusedGrant,
Variables: map[string]interface{}{
"dormant_grants": grantsMap,
"period": criteria.Period.String(),
"retain_duration": criteria.RetainDuration.String(),
"start_date_formatted": startDate.Format("Jan 02, 2006 15:04:05 UTC"),
notifications = append(notifications, domain.Notification{
User: owner,
Labels: map[string]string{
"owner": owner,
"grant_ids": strings.Join(grantIDs, ", "),
},
},
})
}
Message: domain.NotificationMessage{
Type: domain.NotificationTypeUnusedGrant,
Variables: map[string]interface{}{
"dormant_grants": grantsMap,
"period": criteria.Period.String(),
"retain_duration": criteria.RetainDuration.String(),
"start_date_formatted": startDate.Format("Jan 02, 2006 15:04:05 UTC"),
},
},
})
}

if errs := s.notifier.Notify(ctx, notifications); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error(), "provider_urn", provider.URN)
if errs := s.notifier.Notify(ctx, notifications); errs != nil {
for _, err1 := range errs {
s.logger.Error(ctx, "failed to send notifications", "error", err1.Error(), "provider_urn", provider.URN)
}
}
}
}()

return nil
}
Expand Down
Loading