refactor(notification): make notification queue as a service

cmd/wire_gen.go

@@ -56,6 +56,7 @@ import (
 	export2 "github.com/answerdev/answer/internal/service/export"
 	"github.com/answerdev/answer/internal/service/follow"
 	meta2 "github.com/answerdev/answer/internal/service/meta"
+	"github.com/answerdev/answer/internal/service/notice_queue"
 	notification2 "github.com/answerdev/answer/internal/service/notification"
 	"github.com/answerdev/answer/internal/service/notification_common"
 	"github.com/answerdev/answer/internal/service/object_info"
@@ -142,7 +143,8 @@ func initApplication(debug bool, serverConf *conf.Server, dbConf *data.Database,
 	tagCommonService := tag_common2.NewTagCommonService(tagCommonRepo, tagRelRepo, tagRepo, revisionService, siteInfoCommonService)
 	objService := object_info.NewObjService(answerRepo, questionRepo, commentCommonRepo, tagCommonRepo, tagCommonService)
 	voteRepo := activity_common.NewVoteRepo(dataData, activityRepo)
-	commentService := comment2.NewCommentService(commentRepo, commentCommonRepo, userCommon, objService, voteRepo, emailService, userRepo)
+	notificationQueueService := notice_queue.NewNotificationQueueService()
+	commentService := comment2.NewCommentService(commentRepo, commentCommonRepo, userCommon, objService, voteRepo, emailService, userRepo, notificationQueueService)
 	rolePowerRelRepo := role.NewRolePowerRelRepo(dataData)
 	rolePowerRelService := role2.NewRolePowerRelService(rolePowerRelRepo, userRoleRelService)
 	rankService := rank2.NewRankService(userCommon, userRankRepo, objService, userRoleRelService, rolePowerRelService, configService)
@@ -150,7 +152,7 @@ func initApplication(debug bool, serverConf *conf.Server, dbConf *data.Database,
 	reportRepo := report.NewReportRepo(dataData, uniqueIDRepo)
 	reportService := report2.NewReportService(reportRepo, objService)
 	reportController := controller.NewReportController(reportService, rankService)
-	serviceVoteRepo := activity.NewVoteRepo(dataData, uniqueIDRepo, configService, activityRepo, userRankRepo, voteRepo)
+	serviceVoteRepo := activity.NewVoteRepo(dataData, uniqueIDRepo, configService, activityRepo, userRankRepo, voteRepo, notificationQueueService)
 	voteService := service.NewVoteService(serviceVoteRepo, uniqueIDRepo, configService, questionRepo, answerRepo, commentCommonRepo, objService)
 	voteController := controller.NewVoteController(voteService, rankService)
 	followRepo := activity_common.NewFollowRepo(dataData, uniqueIDRepo, activityRepo)
@@ -168,11 +170,11 @@ func initApplication(debug bool, serverConf *conf.Server, dbConf *data.Database,
 	questionCommon := questioncommon.NewQuestionCommon(questionRepo, answerRepo, voteRepo, followRepo, tagCommonService, userCommon, collectionCommon, answerCommon, metaService, configService)
 	collectionService := service.NewCollectionService(collectionRepo, collectionGroupRepo, questionCommon)
 	collectionController := controller.NewCollectionController(collectionService)
-	answerActivityRepo := activity.NewAnswerActivityRepo(dataData, activityRepo, userRankRepo)
+	answerActivityRepo := activity.NewAnswerActivityRepo(dataData, activityRepo, userRankRepo, notificationQueueService)
 	questionActivityRepo := activity.NewQuestionActivityRepo(dataData, activityRepo, userRankRepo)
 	answerActivityService := activity2.NewAnswerActivityService(answerActivityRepo, questionActivityRepo)
-	questionService := service.NewQuestionService(questionRepo, tagCommonService, questionCommon, userCommon, userRepo, revisionService, metaService, collectionCommon, answerActivityService, dataData, emailService)
+	questionService := service.NewQuestionService(questionRepo, tagCommonService, questionCommon, userCommon, userRepo, revisionService, metaService, collectionCommon, answerActivityService, dataData, emailService, notificationQueueService)
-	answerService := service.NewAnswerService(answerRepo, questionRepo, questionCommon, userCommon, collectionCommon, userRepo, revisionService, answerActivityService, answerCommon, voteRepo, emailService, userRoleRelService)
+	answerService := service.NewAnswerService(answerRepo, questionRepo, questionCommon, userCommon, collectionCommon, userRepo, revisionService, answerActivityService, answerCommon, voteRepo, emailService, userRoleRelService, notificationQueueService)
 	questionController := controller.NewQuestionController(questionService, answerService, rankService, siteInfoCommonService)
 	dashboardService := dashboard.NewDashboardService(questionRepo, answerRepo, commentCommonRepo, voteRepo, userRepo, reportRepo, configService, siteInfoCommonService, serviceConf, dataData)
 	answerController := controller.NewAnswerController(answerService, rankService, dashboardService)
@@ -180,10 +182,10 @@ func initApplication(debug bool, serverConf *conf.Server, dbConf *data.Database,
 	searchRepo := search_common.NewSearchRepo(dataData, uniqueIDRepo, userCommon)
 	searchService := service.NewSearchService(searchParser, searchRepo)
 	searchController := controller.NewSearchController(searchService)
-	serviceRevisionService := service.NewRevisionService(revisionRepo, userCommon, questionCommon, answerService, objService, questionRepo, answerRepo, tagRepo, tagCommonService)
+	serviceRevisionService := service.NewRevisionService(revisionRepo, userCommon, questionCommon, answerService, objService, questionRepo, answerRepo, tagRepo, tagCommonService, notificationQueueService)
 	revisionController := controller.NewRevisionController(serviceRevisionService, rankService)
 	rankController := controller.NewRankController(rankService)
-	reportHandle := report_handle_admin.NewReportHandle(questionCommon, commentRepo, configService)
+	reportHandle := report_handle_admin.NewReportHandle(questionCommon, commentRepo, configService, notificationQueueService)
 	reportAdminService := report_admin.NewReportAdminService(reportRepo, userCommon, answerRepo, questionRepo, commentCommonRepo, reportHandle, configService, objService)
 	controller_adminReportController := controller_admin.NewReportController(reportAdminService)
 	userAdminRepo := user.NewUserAdminRepo(dataData, authRepo)
@@ -197,7 +199,7 @@ func initApplication(debug bool, serverConf *conf.Server, dbConf *data.Database,
 	siteInfoController := controller_admin.NewSiteInfoController(siteInfoService)
 	controllerSiteInfoController := controller.NewSiteInfoController(siteInfoCommonService)
 	notificationRepo := notification.NewNotificationRepo(dataData)
-	notificationCommon := notificationcommon.NewNotificationCommon(dataData, notificationRepo, userCommon, activityRepo, followRepo, objService)
+	notificationCommon := notificationcommon.NewNotificationCommon(dataData, notificationRepo, userCommon, activityRepo, followRepo, objService, notificationQueueService)
 	notificationService := notification2.NewNotificationService(dataData, notificationRepo, notificationCommon, revisionService)
 	notificationController := controller.NewNotificationController(notificationService, rankService)
 	dashboardController := controller.NewDashboardController(dashboardService)

internal/repo/activity/answer_repo.go

@@ -19,32 +19,30 @@ import (
 	"xorm.io/xorm"
 )
 
+var (
+	acceptActionList = []string{constant.ActAccept, constant.ActAccepted}
+)
+
 // AnswerActivityRepo answer accepted
 type AnswerActivityRepo struct {
-	data         *data.Data
+	data                     *data.Data
-	activityRepo activity_common.ActivityRepo
+	activityRepo             activity_common.ActivityRepo
-	userRankRepo rank.UserRankRepo
+	userRankRepo             rank.UserRankRepo
+	notificationQueueService notice_queue.NotificationQueueService
 }
 
-const (
-	acceptAction   = "accept"
-	acceptedAction = "accepted"
-)
-
-var (
-	acceptActionList = []string{acceptAction, acceptedAction}
-)
-
 // NewAnswerActivityRepo new repository
 func NewAnswerActivityRepo(
 	data *data.Data,
 	activityRepo activity_common.ActivityRepo,
 	userRankRepo rank.UserRankRepo,
+	notificationQueueService notice_queue.NotificationQueueService,
 ) activity.AnswerActivityRepo {
 	return &AnswerActivityRepo{
-		data:         data,
+		data:                     data,
-		activityRepo: activityRepo,
+		activityRepo:             activityRepo,
-		userRankRepo: userRankRepo,
+		userRankRepo:             userRankRepo,
+		notificationQueueService: notificationQueueService,
 	}
 }
 
@@ -139,7 +137,7 @@ func (ar *AnswerActivityRepo) AcceptAnswer(ctx context.Context,
 			Rank:             deltaRank,
 			HasRank:          hasRank,
 		}
-		if action == acceptAction {
+		if action == constant.ActAccept {
 			addActivity.UserID = questionUserID
 			addActivity.TriggerUserID = converter.StringToInt64(answerUserID)
 			addActivity.OriginalObjectID = questionObjID // if activity is 'accept' means this question is accept the answer.
@@ -209,7 +207,7 @@ func (ar *AnswerActivityRepo) AcceptAnswer(ctx context.Context,
 			msg.ObjectType = constant.AnswerObjectType
 		}
 		if msg.TriggerUserID != msg.ReceiverUserID {
-			notice_queue.AddNotification(msg)
+			ar.notificationQueueService.Send(ctx, msg)
 		}
 	}
 
@@ -223,7 +221,7 @@ func (ar *AnswerActivityRepo) AcceptAnswer(ctx context.Context,
 			msg.TriggerUserID = questionUserID
 			msg.ObjectType = constant.AnswerObjectType
 			msg.NotificationAction = constant.NotificationAcceptAnswer
-			notice_queue.AddNotification(msg)
+			ar.notificationQueueService.Send(ctx, msg)
 		}
 	}
 	return err
@@ -246,7 +244,7 @@ func (ar *AnswerActivityRepo) CancelAcceptAnswer(ctx context.Context,
 			Rank:         -deltaRank,
 			HasRank:      hasRank,
 		}
-		if action == acceptAction {
+		if action == constant.ActAccept {
 			addActivity.UserID = questionUserID
 			addActivity.OriginalObjectID = questionObjID
 		} else {
@@ -303,7 +301,7 @@ func (ar *AnswerActivityRepo) CancelAcceptAnswer(ctx context.Context,
 			msg.ObjectType = constant.AnswerObjectType
 		}
 		if msg.TriggerUserID != msg.ReceiverUserID {
-			notice_queue.AddNotification(msg)
+			ar.notificationQueueService.Send(ctx, msg)
 		}
 	}
 	return err

internal/repo/activity/vote_repo.go

@@ -6,11 +6,11 @@ import (
 	"time"
 
 	"github.com/answerdev/answer/internal/base/constant"
+	"github.com/answerdev/answer/internal/service/notice_queue"
 	"github.com/answerdev/answer/pkg/converter"
 
 	"github.com/answerdev/answer/internal/base/pager"
 	"github.com/answerdev/answer/internal/service/config"
-	"github.com/answerdev/answer/internal/service/notice_queue"
 	"github.com/answerdev/answer/internal/service/rank"
 	"github.com/answerdev/answer/pkg/obj"
 
@@ -30,12 +30,13 @@ import (
 
 // VoteRepo activity repository
 type VoteRepo struct {
-	data          *data.Data
+	data                     *data.Data
-	uniqueIDRepo  unique.UniqueIDRepo
+	uniqueIDRepo             unique.UniqueIDRepo
-	configService *config.ConfigService
+	configService            *config.ConfigService
-	activityRepo  activity_common.ActivityRepo
+	activityRepo             activity_common.ActivityRepo
-	userRankRepo  rank.UserRankRepo
+	userRankRepo             rank.UserRankRepo
-	voteCommon    activity_common.VoteRepo
+	voteCommon               activity_common.VoteRepo
+	notificationQueueService notice_queue.NotificationQueueService
 }
 
 // NewVoteRepo new repository
@@ -46,14 +47,16 @@ func NewVoteRepo(
 	activityRepo activity_common.ActivityRepo,
 	userRankRepo rank.UserRankRepo,
 	voteCommon activity_common.VoteRepo,
+	notificationQueueService notice_queue.NotificationQueueService,
 ) service.VoteRepo {
 	return &VoteRepo{
-		data:          data,
+		data:                     data,
-		uniqueIDRepo:  uniqueIDRepo,
+		uniqueIDRepo:             uniqueIDRepo,
-		configService: configService,
+		configService:            configService,
-		activityRepo:  activityRepo,
+		activityRepo:             activityRepo,
-		userRankRepo:  userRankRepo,
+		userRankRepo:             userRankRepo,
-		voteCommon:    voteCommon,
+		voteCommon:               voteCommon,
+		notificationQueueService: notificationQueueService,
 	}
 }
 
@@ -177,7 +180,7 @@ func (vr *VoteRepo) vote(ctx context.Context, objectID string, userID, objectUse
 		vr.sendNotification(ctx, activityUserID, objectUserID, objectID)
 	}
 	if sendInboxNotification {
-		vr.sendVoteInboxNotification(userID, objectUserID, objectID, upVote)
+		vr.sendVoteInboxNotification(ctx, userID, objectUserID, objectID, upVote)
 	}
 	return
 }
@@ -451,10 +454,10 @@ func (vr *VoteRepo) sendNotification(ctx context.Context, activityUserID, object
 		ObjectID:       objectID,
 		ObjectType:     objectType,
 	}
-	notice_queue.AddNotification(msg)
+	vr.notificationQueueService.Send(ctx, msg)
 }
 
-func (vr *VoteRepo) sendVoteInboxNotification(triggerUserID, receiverUserID, objectID string, upvote bool) {
+func (vr *VoteRepo) sendVoteInboxNotification(ctx context.Context, triggerUserID, receiverUserID, objectID string, upvote bool) {
 	if triggerUserID == receiverUserID {
 		return
 	}
@@ -487,6 +490,6 @@ func (vr *VoteRepo) sendVoteInboxNotification(triggerUserID, receiverUserID, obj
 		}
 	}
 	if len(msg.NotificationAction) > 0 {
-		notice_queue.AddNotification(msg)
+		vr.notificationQueueService.Send(ctx, msg)
 	}
 }

internal/schema/notification_schema.go

@@ -63,6 +63,8 @@ type NotificationMsg struct {
 	NotificationAction string
 	// if true no need to send notification to all followers
 	NoNeedPushAllFollow bool
+	// extra info
+	ExtraInfo map[string]string
 }
 
 type ObjectInfo struct {

internal/service/answer_service.go

@@ -31,18 +31,19 @@ import (
 
 // AnswerService user service
 type AnswerService struct {
-	answerRepo            answercommon.AnswerRepo
+	answerRepo               answercommon.AnswerRepo
-	questionRepo          questioncommon.QuestionRepo
+	questionRepo             questioncommon.QuestionRepo
-	questionCommon        *questioncommon.QuestionCommon
+	questionCommon           *questioncommon.QuestionCommon
-	answerActivityService *activity.AnswerActivityService
+	answerActivityService    *activity.AnswerActivityService
-	userCommon            *usercommon.UserCommon
+	userCommon               *usercommon.UserCommon
-	collectionCommon      *collectioncommon.CollectionCommon
+	collectionCommon         *collectioncommon.CollectionCommon
-	userRepo              usercommon.UserRepo
+	userRepo                 usercommon.UserRepo
-	revisionService       *revision_common.RevisionService
+	revisionService          *revision_common.RevisionService
-	AnswerCommon          *answercommon.AnswerCommon
+	AnswerCommon             *answercommon.AnswerCommon
-	voteRepo              activity_common.VoteRepo
+	voteRepo                 activity_common.VoteRepo
-	emailService          *export.EmailService
+	emailService             *export.EmailService
-	roleService           *role.UserRoleRelService
+	roleService              *role.UserRoleRelService
+	notificationQueueService notice_queue.NotificationQueueService
 }
 
 func NewAnswerService(
@@ -58,20 +59,22 @@ func NewAnswerService(
 	voteRepo activity_common.VoteRepo,
 	emailService *export.EmailService,
 	roleService *role.UserRoleRelService,
+	notificationQueueService notice_queue.NotificationQueueService,
 ) *AnswerService {
 	return &AnswerService{
-		answerRepo:            answerRepo,
+		answerRepo:               answerRepo,
-		questionRepo:          questionRepo,
+		questionRepo:             questionRepo,
-		userCommon:            userCommon,
+		userCommon:               userCommon,
-		collectionCommon:      collectionCommon,
+		collectionCommon:         collectionCommon,
-		questionCommon:        questionCommon,
+		questionCommon:           questionCommon,
-		userRepo:              userRepo,
+		userRepo:                 userRepo,
-		revisionService:       revisionService,
+		revisionService:          revisionService,
-		answerActivityService: answerAcceptActivityRepo,
+		answerActivityService:    answerAcceptActivityRepo,
-		AnswerCommon:          answerCommon,
+		AnswerCommon:             answerCommon,
-		voteRepo:              voteRepo,
+		voteRepo:                 voteRepo,
-		emailService:          emailService,
+		emailService:             emailService,
-		roleService:           roleService,
+		roleService:              roleService,
+		notificationQueueService: notificationQueueService,
 	}
 }
 
@@ -481,7 +484,7 @@ func (as *AnswerService) AdminSetAnswerStatus(ctx context.Context, req *schema.A
 	msg.TriggerUserID = answerInfo.UserID
 	msg.ObjectType = constant.AnswerObjectType
 	msg.NotificationAction = constant.NotificationYourAnswerWasDeleted
-	notice_queue.AddNotification(msg)
+	as.notificationQueueService.Send(ctx, msg)
 
 	return nil
 }
@@ -571,7 +574,7 @@ func (as *AnswerService) notificationUpdateAnswer(ctx context.Context, questionU
 	}
 	msg.ObjectType = constant.AnswerObjectType
 	msg.NotificationAction = constant.NotificationUpdateAnswer
-	notice_queue.AddNotification(msg)
+	as.notificationQueueService.Send(ctx, msg)
 }
 
 func (as *AnswerService) notificationAnswerTheQuestion(ctx context.Context,
@@ -588,7 +591,7 @@ func (as *AnswerService) notificationAnswerTheQuestion(ctx context.Context,
 	}
 	msg.ObjectType = constant.AnswerObjectType
 	msg.NotificationAction = constant.NotificationAnswerTheQuestion
-	notice_queue.AddNotification(msg)
+	as.notificationQueueService.Send(ctx, msg)
 
 	userInfo, exist, err := as.userRepo.GetByUserID(ctx, questionUserID)
 	if err != nil {

internal/service/comment/comment_service.go

@@ -58,13 +58,14 @@ func (c *CommentQuery) GetOrderBy() string {
 
 // CommentService user service
 type CommentService struct {
-	commentRepo       CommentRepo
+	commentRepo              CommentRepo
-	commentCommonRepo comment_common.CommentCommonRepo
+	commentCommonRepo        comment_common.CommentCommonRepo
-	userCommon        *usercommon.UserCommon
+	userCommon               *usercommon.UserCommon
-	voteCommon        activity_common.VoteRepo
+	voteCommon               activity_common.VoteRepo
-	objectInfoService *object_info.ObjService
+	objectInfoService        *object_info.ObjService
-	emailService      *export.EmailService
+	emailService             *export.EmailService
-	userRepo          usercommon.UserRepo
+	userRepo                 usercommon.UserRepo
+	notificationQueueService notice_queue.NotificationQueueService
 }
 
 // NewCommentService new comment service
@@ -76,15 +77,17 @@ func NewCommentService(
 	voteCommon activity_common.VoteRepo,
 	emailService *export.EmailService,
 	userRepo usercommon.UserRepo,
+	notificationQueueService notice_queue.NotificationQueueService,
 ) *CommentService {
 	return &CommentService{
-		commentRepo:       commentRepo,
+		commentRepo:              commentRepo,
-		commentCommonRepo: commentCommonRepo,
+		commentCommonRepo:        commentCommonRepo,
-		userCommon:        userCommon,
+		userCommon:               userCommon,
-		voteCommon:        voteCommon,
+		voteCommon:               voteCommon,
-		objectInfoService: objectInfoService,
+		objectInfoService:        objectInfoService,
-		emailService:      emailService,
+		emailService:             emailService,
-		userRepo:          userRepo,
+		userRepo:                 userRepo,
+		notificationQueueService: notificationQueueService,
 	}
 }
 
@@ -476,7 +479,7 @@ func (cs *CommentService) notificationQuestionComment(ctx context.Context, quest
 	}
 	msg.ObjectType = constant.CommentObjectType
 	msg.NotificationAction = constant.NotificationCommentQuestion
-	notice_queue.AddNotification(msg)
+	cs.notificationQueueService.Send(ctx, msg)
 
 	receiverUserInfo, exist, err := cs.userRepo.GetByUserID(ctx, questionUserID)
 	if err != nil {
@@ -535,7 +538,7 @@ func (cs *CommentService) notificationAnswerComment(ctx context.Context,
 	}
 	msg.ObjectType = constant.CommentObjectType
 	msg.NotificationAction = constant.NotificationCommentAnswer
-	notice_queue.AddNotification(msg)
+	cs.notificationQueueService.Send(ctx, msg)
 
 	receiverUserInfo, exist, err := cs.userRepo.GetByUserID(ctx, answerUserID)
 	if err != nil {
@@ -591,7 +594,7 @@ func (cs *CommentService) notificationCommentReply(ctx context.Context, replyUse
 	}
 	msg.ObjectType = constant.CommentObjectType
 	msg.NotificationAction = constant.NotificationReplyToYou
-	notice_queue.AddNotification(msg)
+	cs.notificationQueueService.Send(ctx, msg)
 }
 
 func (cs *CommentService) notificationMention(
@@ -612,7 +615,7 @@ func (cs *CommentService) notificationMention(
 			}
 			msg.ObjectType = constant.CommentObjectType
 			msg.NotificationAction = constant.NotificationMentionYou
-			notice_queue.AddNotification(msg)
+			cs.notificationQueueService.Send(ctx, msg)
 			alreadyNotifiedUserIDs = append(alreadyNotifiedUserIDs, userInfo.ID)
 		}
 	}

internal/service/notice_queue/notice_queue.go

@@ -1,13 +1,53 @@
 package notice_queue
 
 import (
+	"context"
+
 	"github.com/answerdev/answer/internal/schema"
+	"github.com/segmentfault/pacman/log"
 )
 
-var (
+type NotificationQueueService interface {
-	NotificationQueue = make(chan *schema.NotificationMsg, 128)
+	Send(ctx context.Context, msg *schema.NotificationMsg)
-)
+	RegisterHandler(handler func(ctx context.Context, msg *schema.NotificationMsg) error)
-
+}
-func AddNotification(msg *schema.NotificationMsg) {
+
-	NotificationQueue <- msg
+type notificationQueueService struct {
+	Queue   chan *schema.NotificationMsg
+	Handler func(ctx context.Context, msg *schema.NotificationMsg) error
+}
+
+func (ns *notificationQueueService) Send(ctx context.Context, msg *schema.NotificationMsg) {
+	ns.Queue <- msg
+}
+
+func (ns *notificationQueueService) RegisterHandler(
+	handler func(ctx context.Context, msg *schema.NotificationMsg) error) {
+	ns.Handler = handler
+}
+
+func (ns *notificationQueueService) working() {
+	go func() {
+		for msg := range ns.Queue {
+			log.Debugf("received notification %+v", msg)
+			if ns.Handler == nil {
+				log.Warnf("no handler for notification")
+				continue
+			}
+			if err := ns.Handler(context.Background(), msg); err != nil {
+				log.Error(err)
+			}
+		}
+	}()
+}
+
+// NewNotificationQueueService create a new notification queue service
+func NewNotificationQueueService() NotificationQueueService {
+	ns := &notificationQueueService{}
+	ns.Queue = make(chan *schema.NotificationMsg, 128)
+	ns.working()
+	return ns
+}
+
+func AddNotification2(msg *schema.NotificationMsg) {
 }

internal/service/notification_common/notification.go

@@ -33,12 +33,13 @@ type NotificationRepo interface {
 }
 
 type NotificationCommon struct {
-	data              *data.Data
+	data                     *data.Data
-	notificationRepo  NotificationRepo
+	notificationRepo         NotificationRepo
-	activityRepo      activity_common.ActivityRepo
+	activityRepo             activity_common.ActivityRepo
-	followRepo        activity_common.FollowRepo
+	followRepo               activity_common.FollowRepo
-	userCommon        *usercommon.UserCommon
+	userCommon               *usercommon.UserCommon
-	objectInfoService *object_info.ObjService
+	objectInfoService        *object_info.ObjService
+	notificationQueueService notice_queue.NotificationQueueService
 }
 
 func NewNotificationCommon(
@@ -48,31 +49,21 @@ func NewNotificationCommon(
 	activityRepo activity_common.ActivityRepo,
 	followRepo activity_common.FollowRepo,
 	objectInfoService *object_info.ObjService,
+	notificationQueueService notice_queue.NotificationQueueService,
 ) *NotificationCommon {
 	notification := &NotificationCommon{
-		data:              data,
+		data:                     data,
-		notificationRepo:  notificationRepo,
+		notificationRepo:         notificationRepo,
-		activityRepo:      activityRepo,
+		activityRepo:             activityRepo,
-		followRepo:        followRepo,
+		followRepo:               followRepo,
-		userCommon:        userCommon,
+		userCommon:               userCommon,
-		objectInfoService: objectInfoService,
+		objectInfoService:        objectInfoService,
+		notificationQueueService: notificationQueueService,
 	}
-	notification.HandleNotification()
+	notificationQueueService.RegisterHandler(notification.AddNotification)
 	return notification
 }
 
-func (ns *NotificationCommon) HandleNotification() {
-	go func() {
-		for msg := range notice_queue.NotificationQueue {
-			log.Debugf("received notification %+v", msg)
-			err := ns.AddNotification(context.TODO(), msg)
-			if err != nil {
-				log.Error(err)
-			}
-		}
-	}()
-}
-
 // AddNotification
 // need set
 // LoginUserID
@@ -213,6 +204,6 @@ func (ns *NotificationCommon) SendNotificationToAllFollower(ctx context.Context,
 		t.ReceiverUserID = userID
 		t.TriggerUserID = msg.TriggerUserID
 		t.NoNeedPushAllFollow = true
-		notice_queue.AddNotification(t)
+		ns.notificationQueueService.Send(ctx, t)
 	}
 }

internal/service/provider.go

@@ -14,6 +14,7 @@ import (
 	"github.com/answerdev/answer/internal/service/export"
 	"github.com/answerdev/answer/internal/service/follow"
 	"github.com/answerdev/answer/internal/service/meta"
+	"github.com/answerdev/answer/internal/service/notice_queue"
 	"github.com/answerdev/answer/internal/service/notification"
 	notficationcommon "github.com/answerdev/answer/internal/service/notification_common"
 	"github.com/answerdev/answer/internal/service/object_info"
@@ -86,4 +87,5 @@ var ProviderSetService = wire.NewSet(
 	user_external_login.NewUserCenterLoginService,
 	plugin_common.NewPluginCommonService,
 	config.NewConfigService,
+	notice_queue.NewNotificationQueueService,
 )

internal/service/question_service.go

@@ -42,17 +42,18 @@ import (
 
 // QuestionService user service
 type QuestionService struct {
-	questionRepo          questioncommon.QuestionRepo
+	questionRepo             questioncommon.QuestionRepo
-	tagCommon             *tagcommon.TagCommonService
+	tagCommon                *tagcommon.TagCommonService
-	questioncommon        *questioncommon.QuestionCommon
+	questioncommon           *questioncommon.QuestionCommon
-	userCommon            *usercommon.UserCommon
+	userCommon               *usercommon.UserCommon
-	userRepo              usercommon.UserRepo
+	userRepo                 usercommon.UserRepo
-	revisionService       *revision_common.RevisionService
+	revisionService          *revision_common.RevisionService
-	metaService           *meta.MetaService
+	metaService              *meta.MetaService
-	collectionCommon      *collectioncommon.CollectionCommon
+	collectionCommon         *collectioncommon.CollectionCommon
-	answerActivityService *activity.AnswerActivityService
+	answerActivityService    *activity.AnswerActivityService
-	data                  *data.Data
+	data                     *data.Data
-	emailService          *export.EmailService
+	emailService             *export.EmailService
+	notificationQueueService notice_queue.NotificationQueueService
 }
 
 func NewQuestionService(
@@ -67,19 +68,21 @@ func NewQuestionService(
 	answerActivityService *activity.AnswerActivityService,
 	data *data.Data,
 	emailService *export.EmailService,
+	notificationQueueService notice_queue.NotificationQueueService,
 ) *QuestionService {
 	return &QuestionService{
-		questionRepo:          questionRepo,
+		questionRepo:             questionRepo,
-		tagCommon:             tagCommon,
+		tagCommon:                tagCommon,
-		questioncommon:        questioncommon,
+		questioncommon:           questioncommon,
-		userCommon:            userCommon,
+		userCommon:               userCommon,
-		userRepo:              userRepo,
+		userRepo:                 userRepo,
-		revisionService:       revisionService,
+		revisionService:          revisionService,
-		metaService:           metaService,
+		metaService:              metaService,
-		collectionCommon:      collectionCommon,
+		collectionCommon:         collectionCommon,
-		answerActivityService: answerActivityService,
+		answerActivityService:    answerActivityService,
-		data:                  data,
+		data:                     data,
-		emailService:          emailService,
+		emailService:             emailService,
+		notificationQueueService: notificationQueueService,
 	}
 }
 
@@ -633,7 +636,7 @@ func (qs *QuestionService) notificationInviteUser(
 		}
 		msg.ObjectType = constant.QuestionObjectType
 		msg.NotificationAction = constant.NotificationInvitedYouToAnswer
-		notice_queue.AddNotification(msg)
+		qs.notificationQueueService.Send(ctx, msg)
 
 		userInfo, ok := invitee[userID]
 		if !ok {
@@ -1247,7 +1250,7 @@ func (qs *QuestionService) AdminSetQuestionStatus(ctx context.Context, questionI
 	msg.TriggerUserID = questionInfo.UserID
 	msg.ObjectType = constant.QuestionObjectType
 	msg.NotificationAction = constant.NotificationYourQuestionWasDeleted
-	notice_queue.AddNotification(msg)
+	qs.notificationQueueService.Send(ctx, msg)
 	return nil
 }
 

internal/service/report_handle_admin/report_handle.go

@@ -4,30 +4,34 @@ import (
 	"context"
 
 	"github.com/answerdev/answer/internal/service/config"
+	"github.com/answerdev/answer/internal/service/notice_queue"
 
 	"github.com/answerdev/answer/internal/base/constant"
 	"github.com/answerdev/answer/internal/entity"
 	"github.com/answerdev/answer/internal/schema"
 	"github.com/answerdev/answer/internal/service/comment"
-	"github.com/answerdev/answer/internal/service/notice_queue"
 	questioncommon "github.com/answerdev/answer/internal/service/question_common"
 	"github.com/answerdev/answer/pkg/obj"
 )
 
 type ReportHandle struct {
-	questionCommon *questioncommon.QuestionCommon
+	questionCommon           *questioncommon.QuestionCommon
-	commentRepo    comment.CommentRepo
+	commentRepo              comment.CommentRepo
-	configService  *config.ConfigService
+	configService            *config.ConfigService
+	notificationQueueService notice_queue.NotificationQueueService
 }
 
 func NewReportHandle(
 	questionCommon *questioncommon.QuestionCommon,
 	commentRepo comment.CommentRepo,
-	configService *config.ConfigService) *ReportHandle {
+	configService *config.ConfigService,
+	notificationQueueService notice_queue.NotificationQueueService,
+) *ReportHandle {
 	return &ReportHandle{
-		questionCommon: questionCommon,
+		questionCommon:           questionCommon,
-		commentRepo:    commentRepo,
+		commentRepo:              commentRepo,
-		configService:  configService,
+		configService:            configService,
+		notificationQueueService: notificationQueueService,
 	}
 }
 
@@ -88,5 +92,5 @@ func (rh *ReportHandle) sendNotification(ctx context.Context, reportedUserID, ob
 		ObjectType:         constant.ReportObjectType,
 		NotificationAction: notificationAction,
 	}
-	notice_queue.AddNotification(msg)
+	rh.notificationQueueService.Send(ctx, msg)
 }

internal/service/revision_service.go

@@ -28,15 +28,16 @@ import (
 
 // RevisionService user service
 type RevisionService struct {
-	revisionRepo      revision.RevisionRepo
+	revisionRepo             revision.RevisionRepo
-	userCommon        *usercommon.UserCommon
+	userCommon               *usercommon.UserCommon
-	questionCommon    *questioncommon.QuestionCommon
+	questionCommon           *questioncommon.QuestionCommon
-	answerService     *AnswerService
+	answerService            *AnswerService
-	objectInfoService *object_info.ObjService
+	objectInfoService        *object_info.ObjService
-	questionRepo      questioncommon.QuestionRepo
+	questionRepo             questioncommon.QuestionRepo
-	answerRepo        answercommon.AnswerRepo
+	answerRepo               answercommon.AnswerRepo
-	tagRepo           tag_common.TagRepo
+	tagRepo                  tag_common.TagRepo
-	tagCommon         *tagcommon.TagCommonService
+	tagCommon                *tagcommon.TagCommonService
+	notificationQueueService notice_queue.NotificationQueueService
 }
 
 func NewRevisionService(
@@ -49,17 +50,19 @@ func NewRevisionService(
 	answerRepo answercommon.AnswerRepo,
 	tagRepo tag_common.TagRepo,
 	tagCommon *tagcommon.TagCommonService,
+	notificationQueueService notice_queue.NotificationQueueService,
 ) *RevisionService {
 	return &RevisionService{
-		revisionRepo:      revisionRepo,
+		revisionRepo:             revisionRepo,
-		userCommon:        userCommon,
+		userCommon:               userCommon,
-		questionCommon:    questionCommon,
+		questionCommon:           questionCommon,
-		answerService:     answerService,
+		answerService:            answerService,
-		objectInfoService: objectInfoService,
+		objectInfoService:        objectInfoService,
-		questionRepo:      questionRepo,
+		questionRepo:             questionRepo,
-		answerRepo:        answerRepo,
+		answerRepo:               answerRepo,
-		tagRepo:           tagRepo,
+		tagRepo:                  tagRepo,
-		tagCommon:         tagCommon,
+		tagCommon:                tagCommon,
+		notificationQueueService: notificationQueueService,
 	}
 }
 
@@ -210,7 +213,7 @@ func (rs *RevisionService) revisionAuditAnswer(ctx context.Context, revisionitem
 		}
 		msg.ObjectType = constant.AnswerObjectType
 		msg.NotificationAction = constant.NotificationUpdateAnswer
-		notice_queue.AddNotification(msg)
+		rs.notificationQueueService.Send(ctx, msg)
 
 		activity_queue.AddActivity(&schema.ActivityMsg{
 			UserID:           revisionitem.UserID,