diff --git a/internal/repo/notification/notification_repo.go b/internal/repo/notification/notification_repo.go index 8d8e707f2..ec626d531 100644 --- a/internal/repo/notification/notification_repo.go +++ b/internal/repo/notification/notification_repo.go @@ -76,12 +76,20 @@ func (nr *notificationRepo) ClearUnRead(ctx context.Context, userID string, noti return } -func (nr *notificationRepo) ClearIDUnRead(ctx context.Context, userID string, id string) (err error) { +// ClearIDUnRead marks the notification as read only when it is still unread. It returns the +// number of changed rows, so the caller can tell whether the notification really changed from +// unread to read and avoid decreasing the red dot cache for duplicate requests. +func (nr *notificationRepo) ClearIDUnRead(ctx context.Context, userID string, id string) (affected int64, err error) { info := &entity.Notification{} info.IsRead = schema.NotificationRead - _, err = nr.data.DB.Context(ctx).Where("user_id = ?", userID).And("id = ?", id).Cols("is_read").Update(info) + affected, err = nr.data.DB.Context(ctx). + Where("user_id = ?", userID). + And("id = ?", id). + And("is_read = ?", schema.NotificationNotRead). + Cols("is_read"). + Update(info) if err != nil { - return errors.InternalServer(reason.DatabaseError).WithError(err).WithStack() + err = errors.InternalServer(reason.DatabaseError).WithError(err).WithStack() } return } diff --git a/internal/repo/repo_test/notification_repo_test.go b/internal/repo/repo_test/notification_repo_test.go index 4f843e777..9bd640d99 100644 --- a/internal/repo/repo_test/notification_repo_test.go +++ b/internal/repo/repo_test/notification_repo_test.go @@ -21,6 +21,7 @@ package repo_test import ( "context" + "sync" "testing" "github.com/apache/answer/internal/entity" @@ -47,13 +48,60 @@ func Test_notificationRepo_ClearIDUnRead(t *testing.T) { err := notificationRepo.AddNotification(context.TODO(), ent) require.NoError(t, err) - err = notificationRepo.ClearIDUnRead(context.TODO(), ent.UserID, ent.ID) + affected, err := notificationRepo.ClearIDUnRead(context.TODO(), ent.UserID, ent.ID) require.NoError(t, err) + assert.EqualValues(t, 1, affected) got, exists, err := notificationRepo.GetById(context.TODO(), ent.ID) require.NoError(t, err) assert.True(t, exists) assert.Equal(t, schema.NotificationRead, got.IsRead) + + // a duplicate request must not report another unread to read transition + affected, err = notificationRepo.ClearIDUnRead(context.TODO(), ent.UserID, ent.ID) + require.NoError(t, err) + assert.EqualValues(t, 0, affected) + + // a notification of another user must not be changed + affected, err = notificationRepo.ClearIDUnRead(context.TODO(), "another-user", ent.ID) + require.NoError(t, err) + assert.EqualValues(t, 0, affected) +} + +func Test_notificationRepo_ClearIDUnReadConcurrent(t *testing.T) { + notificationRepo := notification.NewNotificationRepo(testDataSource) + ent := buildNotificationEntity() + err := notificationRepo.AddNotification(context.TODO(), ent) + require.NoError(t, err) + + const requestCount = 5 + var ( + waitGroup sync.WaitGroup + mutex sync.Mutex + affected int64 + errs []error + ) + start := make(chan struct{}) + for i := 0; i < requestCount; i++ { + waitGroup.Add(1) + go func() { + defer waitGroup.Done() + <-start + rows, err := notificationRepo.ClearIDUnRead(context.TODO(), ent.UserID, ent.ID) + mutex.Lock() + defer mutex.Unlock() + if err != nil { + errs = append(errs, err) + return + } + affected += rows + }() + } + close(start) + waitGroup.Wait() + + require.Empty(t, errs) + assert.EqualValues(t, 1, affected, "only one concurrent request may change the notification") } func Test_notificationRepo_ClearUnRead(t *testing.T) { diff --git a/internal/service/notification/notification_service.go b/internal/service/notification/notification_service.go index 6a69cbaef..2cd9cfd7f 100644 --- a/internal/service/notification/notification_service.go +++ b/internal/service/notification/notification_service.go @@ -190,11 +190,12 @@ func (ns *NotificationService) ClearIDUnRead(ctx context.Context, userID string, if !exist || notificationInfo.UserID != userID { return nil } - if notificationInfo.IsRead == schema.NotificationNotRead { - err := ns.notificationRepo.ClearIDUnRead(ctx, userID, id) - if err != nil { - return err - } + + // The update is conditional and atomic, it only changes the notification while it is still + // unread and reports how many rows were changed. + affected, err := ns.notificationRepo.ClearIDUnRead(ctx, userID, id) + if err != nil { + return err } err = ns.notificationCommon.RemoveBadgeAwardAlertCache(ctx, userID, id) @@ -202,6 +203,11 @@ func (ns *NotificationService) ClearIDUnRead(ctx context.Context, userID string, log.Errorf("remove badge award alert cache failed: %v", err) } + if affected == 0 { + // The notification was already read, so the red dot cache must not be decreased again. + return nil + } + _ = ns.notificationCommon.DecreaseRedDot(ctx, userID, notificationInfo.Type) return nil } diff --git a/internal/service/notification/notification_service_test.go b/internal/service/notification/notification_service_test.go new file mode 100644 index 000000000..2c3a95ad5 --- /dev/null +++ b/internal/service/notification/notification_service_test.go @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package notification + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/apache/answer/internal/base/constant" + basedata "github.com/apache/answer/internal/base/data" + "github.com/apache/answer/internal/entity" + "github.com/apache/answer/internal/schema" + "github.com/apache/answer/internal/service/noticequeue" + notficationcommon "github.com/apache/answer/internal/service/notification_common" + "github.com/segmentfault/pacman/cache" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// redDotTestNotificationRepo is an in-memory notification repo. Its ClearIDUnRead only changes +// the notification while it is still unread and reports the changed rows, the same way the +// database repo does. +type redDotTestNotificationRepo struct { + notifications map[string]*entity.Notification +} + +func (f *redDotTestNotificationRepo) AddNotification(_ context.Context, notification *entity.Notification) error { + f.notifications[notification.ID] = notification + return nil +} + +func (f *redDotTestNotificationRepo) GetNotificationPage(_ context.Context, _ *schema.NotificationSearch) ( + []*entity.Notification, int64, error) { + return nil, 0, nil +} + +func (f *redDotTestNotificationRepo) ClearUnRead(_ context.Context, _ string, _ int) error { + return nil +} + +func (f *redDotTestNotificationRepo) ClearIDUnRead(_ context.Context, userID string, id string) (int64, error) { + notification, exist := f.notifications[id] + if !exist || notification.UserID != userID || notification.IsRead != schema.NotificationNotRead { + return 0, nil + } + notification.IsRead = schema.NotificationRead + return 1, nil +} + +func (f *redDotTestNotificationRepo) GetByUserIdObjectIdTypeId(_ context.Context, _, _ string, _ int) ( + *entity.Notification, bool, error) { + return nil, false, nil +} + +func (f *redDotTestNotificationRepo) UpdateNotificationContent(_ context.Context, _ *entity.Notification) error { + return nil +} + +func (f *redDotTestNotificationRepo) GetById(_ context.Context, id string) (*entity.Notification, bool, error) { + notification, exist := f.notifications[id] + return notification, exist, nil +} + +func (f *redDotTestNotificationRepo) CountNotificationByUser(_ context.Context, _ *entity.Notification) (int64, error) { + return 0, nil +} + +func (f *redDotTestNotificationRepo) DeleteNotification(_ context.Context, _ string) error { + return nil +} + +func (f *redDotTestNotificationRepo) DeleteUserNotificationConfig(_ context.Context, _ string) error { + return nil +} + +// TestNotificationService_ClearIDUnReadDecreaseRedDotOnce checks that reading the same +// notification more than once only decreases the red dot cache once, because only the request +// that really changes the notification from unread to read may decrease it. +func TestNotificationService_ClearIDUnReadDecreaseRedDotOnce(t *testing.T) { + ctx := context.Background() + const ( + userID = "red-dot-user" + notificationID = "100" + ) + + cacheStore, cleanup, err := basedata.NewCache(&basedata.CacheConf{}) + require.NoError(t, err) + t.Cleanup(cleanup) + + testData := &basedata.Data{Cache: cacheStore} + repo := &redDotTestNotificationRepo{notifications: map[string]*entity.Notification{}} + info := &entity.Notification{ + ID: notificationID, + UserID: userID, + ObjectID: "1", + Content: "{}", + Type: schema.NotificationTypeInbox, + IsRead: schema.NotificationNotRead, + Status: schema.NotificationStatusNormal, + } + require.NoError(t, repo.AddNotification(ctx, info)) + + service := &NotificationService{ + data: testData, + notificationRepo: repo, + notificationCommon: notficationcommon.NewNotificationCommon( + testData, repo, nil, nil, nil, nil, noticequeue.NewService(), nil, nil), + } + + // the user has 3 unread notifications and reads one of them + redDotKey := fmt.Sprintf(constant.RedDotCacheKey, constant.NotificationTypeInbox, userID) + require.NoError(t, cacheStore.SetInt64(ctx, redDotKey, 3, time.Hour)) + + require.NoError(t, service.ClearIDUnRead(ctx, userID, notificationID)) + assertRedDotCount(t, ctx, cacheStore, redDotKey, 2) + + // a duplicate request must not decrease the red dot cache again + require.NoError(t, service.ClearIDUnRead(ctx, userID, notificationID)) + assertRedDotCount(t, ctx, cacheStore, redDotKey, 2) + + // another user can neither read the notification nor change the red dot cache + require.NoError(t, service.ClearIDUnRead(ctx, "another-user", notificationID)) + assertRedDotCount(t, ctx, cacheStore, redDotKey, 2) +} + +func assertRedDotCount(t *testing.T, ctx context.Context, c cache.Cache, key string, expected int64) { + count, exist, err := c.GetInt64(ctx, key) + require.NoError(t, err) + require.True(t, exist, "red dot cache key should exist") + assert.Equal(t, expected, count) +} diff --git a/internal/service/notification_common/notification.go b/internal/service/notification_common/notification.go index aa3f4106c..1edecb285 100644 --- a/internal/service/notification_common/notification.go +++ b/internal/service/notification_common/notification.go @@ -50,7 +50,7 @@ type NotificationRepo interface { AddNotification(ctx context.Context, notification *entity.Notification) (err error) GetNotificationPage(ctx context.Context, search *schema.NotificationSearch) ([]*entity.Notification, int64, error) ClearUnRead(ctx context.Context, userID string, notificationType int) (err error) - ClearIDUnRead(ctx context.Context, userID string, id string) (err error) + ClearIDUnRead(ctx context.Context, userID string, id string) (affected int64, err error) GetByUserIdObjectIdTypeId(ctx context.Context, userID, objectID string, notificationType int) (*entity.Notification, bool, error) UpdateNotificationContent(ctx context.Context, notification *entity.Notification) (err error) GetById(ctx context.Context, id string) (*entity.Notification, bool, error)