wehub-resource-sync
324 lines
11 KiB
Go
324 lines
11 KiB
Go
/*
|
|
* 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"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/apache/answer/internal/base/constant"
|
|
"github.com/apache/answer/internal/base/translator"
|
|
"github.com/apache/answer/internal/schema"
|
|
"github.com/apache/answer/pkg/display"
|
|
"github.com/apache/answer/plugin"
|
|
"github.com/jinzhu/copier"
|
|
"github.com/segmentfault/pacman/i18n"
|
|
"github.com/segmentfault/pacman/log"
|
|
)
|
|
|
|
type NewQuestionSubscriber struct {
|
|
UserID string `json:"user_id"`
|
|
Channels schema.NotificationChannels `json:"channels"`
|
|
NotificationSource constant.NotificationSource `json:"notification_source"`
|
|
}
|
|
|
|
func (ns *ExternalNotificationService) handleNewQuestionNotification(ctx context.Context,
|
|
msg *schema.ExternalNotificationMsg) error {
|
|
log.Debugf("try to send new question notification %+v", msg)
|
|
subscribers, err := ns.getNewQuestionSubscribers(ctx, msg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
log.Debugf("get subscribers %d for question %s", len(subscribers), msg.NewQuestionTemplateRawData.QuestionID)
|
|
|
|
ns.syncNewQuestionNotificationToPlugin(ctx, msg)
|
|
ns.enqueueNewQuestionNotificationEmails(subscribers, msg.NewQuestionTemplateRawData)
|
|
return nil
|
|
}
|
|
|
|
func (ns *ExternalNotificationService) enqueueNewQuestionNotificationEmails(
|
|
subscribers []*NewQuestionSubscriber,
|
|
rawData *schema.NewQuestionTemplateRawData,
|
|
) {
|
|
task := newQuestionEmailTaskFromRawData(collectNewQuestionNotificationEmailUserIDs(subscribers), rawData)
|
|
if len(task.UserIDs) == 0 {
|
|
return
|
|
}
|
|
if ns.newQuestionEmailWorker == nil {
|
|
log.Warnf("[new_question_email] worker is nil, dropping task for question %s", task.QuestionID)
|
|
return
|
|
}
|
|
if !ns.newQuestionEmailWorker.TryEnqueue(task) {
|
|
log.Warnf("[new_question_email] failed to enqueue task for question %s", task.QuestionID)
|
|
}
|
|
}
|
|
|
|
func collectNewQuestionNotificationEmailUserIDs(subscribers []*NewQuestionSubscriber) []string {
|
|
userIDs := make([]string, 0, len(subscribers))
|
|
for _, subscriber := range subscribers {
|
|
if subscriber == nil {
|
|
continue
|
|
}
|
|
for _, channel := range subscriber.Channels {
|
|
if channel == nil || !channel.Enable || channel.Key != constant.EmailChannel {
|
|
continue
|
|
}
|
|
userIDs = append(userIDs, subscriber.UserID)
|
|
}
|
|
}
|
|
return userIDs
|
|
}
|
|
|
|
func (ns *ExternalNotificationService) getNewQuestionSubscribers(ctx context.Context, msg *schema.ExternalNotificationMsg) (
|
|
subscribers []*NewQuestionSubscriber, err error) {
|
|
subscribersMapping := make(map[string]*NewQuestionSubscriber)
|
|
|
|
// 1. get all this new question's tags followers
|
|
tagsFollowerIDs := make([]string, 0)
|
|
followerMapping := make(map[string]bool)
|
|
for _, tagID := range msg.NewQuestionTemplateRawData.TagIDs {
|
|
userIDs, err := ns.followRepo.GetFollowUserIDs(ctx, tagID)
|
|
if err != nil {
|
|
log.Error(err)
|
|
continue
|
|
}
|
|
for _, userID := range userIDs {
|
|
if _, ok := followerMapping[userID]; ok {
|
|
continue
|
|
}
|
|
followerMapping[userID] = true
|
|
tagsFollowerIDs = append(tagsFollowerIDs, userID)
|
|
}
|
|
}
|
|
userNotificationConfigs, err := ns.userNotificationConfigRepo.GetByUsersAndSource(
|
|
ctx, tagsFollowerIDs, constant.AllNewQuestionForFollowingTagsSource)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, userNotificationConfig := range userNotificationConfigs {
|
|
if _, ok := subscribersMapping[userNotificationConfig.UserID]; ok {
|
|
continue
|
|
}
|
|
subscribersMapping[userNotificationConfig.UserID] = &NewQuestionSubscriber{
|
|
UserID: userNotificationConfig.UserID,
|
|
Channels: schema.NewNotificationChannelsFormJson(userNotificationConfig.Channels),
|
|
NotificationSource: constant.AllNewQuestionForFollowingTagsSource,
|
|
}
|
|
}
|
|
log.Debugf("get %d subscribers from tags", len(subscribersMapping))
|
|
|
|
// 2. get all new question's followers
|
|
notificationConfigs, err := ns.userNotificationConfigRepo.GetBySource(ctx, constant.AllNewQuestionSource)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, notificationConfig := range notificationConfigs {
|
|
if _, ok := subscribersMapping[notificationConfig.UserID]; ok {
|
|
continue
|
|
}
|
|
if ns.checkSendNewQuestionNotificationEmailLimit(ctx, notificationConfig.UserID) {
|
|
continue
|
|
}
|
|
subscribersMapping[notificationConfig.UserID] = &NewQuestionSubscriber{
|
|
UserID: notificationConfig.UserID,
|
|
Channels: schema.NewNotificationChannelsFormJson(notificationConfig.Channels),
|
|
NotificationSource: constant.AllNewQuestionSource,
|
|
}
|
|
}
|
|
|
|
// 3. remove question owner
|
|
delete(subscribersMapping, msg.NewQuestionTemplateRawData.QuestionAuthorUserID)
|
|
for _, subscriber := range subscribersMapping {
|
|
subscribers = append(subscribers, subscriber)
|
|
}
|
|
log.Debugf("get %d subscribers from all new question config", len(subscribers))
|
|
return subscribers, nil
|
|
}
|
|
|
|
func (ns *ExternalNotificationService) checkSendNewQuestionNotificationEmailLimit(ctx context.Context, userID string) bool {
|
|
key := constant.NewQuestionNotificationLimitCacheKeyPrefix + userID
|
|
old, exist, err := ns.data.Cache.GetInt64(ctx, key)
|
|
if err != nil {
|
|
log.Error(err)
|
|
return false
|
|
}
|
|
if exist && old >= constant.NewQuestionNotificationLimitMax {
|
|
log.Debugf("%s user reach new question notification limit", userID)
|
|
return true
|
|
}
|
|
if !exist {
|
|
err = ns.data.Cache.SetInt64(ctx, key, 1, constant.NewQuestionNotificationLimitCacheTime)
|
|
} else {
|
|
_, err = ns.data.Cache.Increase(ctx, key, 1)
|
|
}
|
|
if err != nil {
|
|
log.Error(err)
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (ns *ExternalNotificationService) sendNewQuestionNotificationEmail(ctx context.Context,
|
|
userID string, rawData *schema.NewQuestionTemplateRawData) {
|
|
if unavailable := ns.checkUserStatusBeforeNotification(ctx, userID); unavailable {
|
|
return
|
|
}
|
|
userInfo, exist, err := ns.userRepo.GetByUserID(ctx, userID)
|
|
if err != nil {
|
|
log.Error(err)
|
|
return
|
|
}
|
|
if !exist {
|
|
log.Errorf("user %s not exist", userID)
|
|
return
|
|
}
|
|
// If receiver has set language, use it to send email.
|
|
if len(userInfo.Language) > 0 {
|
|
ctx = context.WithValue(ctx, constant.AcceptLanguageContextKey, i18n.Language(userInfo.Language))
|
|
}
|
|
title, body, err := ns.emailService.NewQuestionTemplate(ctx, rawData)
|
|
if err != nil {
|
|
log.Error(err)
|
|
return
|
|
}
|
|
|
|
codeContent := &schema.EmailCodeContent{
|
|
SourceType: schema.UnsubscribeSourceType,
|
|
Email: userInfo.EMail,
|
|
UserID: userID,
|
|
NotificationSources: []constant.NotificationSource{
|
|
constant.AllNewQuestionSource,
|
|
constant.AllNewQuestionForFollowingTagsSource,
|
|
},
|
|
SkipValidationLatestCode: true,
|
|
}
|
|
ns.emailService.SendAndSaveCodeWithTime(
|
|
ctx, userInfo.ID, userInfo.EMail, title, body, rawData.UnsubscribeCode, codeContent.ToJSONString(), 1*24*time.Hour)
|
|
}
|
|
|
|
func (ns *ExternalNotificationService) syncNewQuestionNotificationToPlugin(ctx context.Context,
|
|
msg *schema.ExternalNotificationMsg) {
|
|
_ = plugin.CallNotification(func(fn plugin.Notification) error {
|
|
// 1. get all this new question's tags followers
|
|
subscribersMapping := make(map[string]plugin.NotificationType)
|
|
for _, tagID := range msg.NewQuestionTemplateRawData.TagIDs {
|
|
userIDs, err := ns.followRepo.GetFollowUserIDs(ctx, tagID)
|
|
if err != nil {
|
|
log.Error(err)
|
|
continue
|
|
}
|
|
for _, userID := range userIDs {
|
|
subscribersMapping[userID] = plugin.NotificationNewQuestionFollowedTag
|
|
}
|
|
}
|
|
|
|
// 2. get all new question's followers
|
|
questionSubscribers := fn.GetNewQuestionSubscribers()
|
|
for _, subscriber := range questionSubscribers {
|
|
subscribersMapping[subscriber] = plugin.NotificationNewQuestion
|
|
}
|
|
|
|
// 3. remove question owner
|
|
delete(subscribersMapping, msg.NewQuestionTemplateRawData.QuestionAuthorUserID)
|
|
|
|
pluginNotificationMsg := ns.newPluginQuestionNotification(ctx, msg)
|
|
|
|
// 4. send notification
|
|
for subscriberUserID, notificationType := range subscribersMapping {
|
|
newMsg := plugin.NotificationMessage{}
|
|
_ = copier.Copy(&newMsg, pluginNotificationMsg)
|
|
newMsg.ReceiverUserID = subscriberUserID
|
|
newMsg.Type = notificationType
|
|
|
|
if len(subscriberUserID) > 0 {
|
|
userInfo, _, _ := ns.userRepo.GetByUserID(ctx, subscriberUserID)
|
|
if userInfo != nil && len(userInfo.Language) > 0 && userInfo.Language != translator.DefaultLangOption {
|
|
newMsg.ReceiverLang = userInfo.Language
|
|
}
|
|
}
|
|
|
|
// Get all external logins as fallback
|
|
externalLogins, err := ns.userExternalLoginRepo.GetUserExternalLoginList(ctx, subscriberUserID)
|
|
if err != nil {
|
|
log.Errorf("get user external login list failed for user %s: %v", subscriberUserID, err)
|
|
} else if len(externalLogins) > 0 {
|
|
newMsg.ReceiverExternalID = externalLogins[0].ExternalID
|
|
if len(externalLogins) > 1 {
|
|
log.Debugf("user %s has %d SSO logins, using most recent: provider=%s",
|
|
subscriberUserID, len(externalLogins), externalLogins[0].Provider)
|
|
}
|
|
}
|
|
|
|
// Try to get external login specific to this plugin (takes precedence over fallback)
|
|
userInfo, exist, err := ns.userExternalLoginRepo.GetByUserID(ctx, fn.Info().SlugName, subscriberUserID)
|
|
if err != nil {
|
|
log.Errorf("get user external login info failed: %v", err)
|
|
return nil
|
|
}
|
|
if exist {
|
|
newMsg.ReceiverExternalID = userInfo.ExternalID
|
|
}
|
|
fn.Notify(newMsg)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
func (ns *ExternalNotificationService) newPluginQuestionNotification(
|
|
ctx context.Context, msg *schema.ExternalNotificationMsg) (raw *plugin.NotificationMessage) {
|
|
raw = &plugin.NotificationMessage{
|
|
ReceiverUserID: msg.ReceiverUserID,
|
|
ReceiverLang: msg.ReceiverLang,
|
|
QuestionTitle: msg.NewQuestionTemplateRawData.QuestionTitle,
|
|
QuestionTags: strings.Join(msg.NewQuestionTemplateRawData.Tags, ","),
|
|
}
|
|
siteInfo, err := ns.siteInfoService.GetSiteGeneral(ctx)
|
|
if err != nil {
|
|
return raw
|
|
}
|
|
seoInfo, err := ns.siteInfoService.GetSiteSeo(ctx)
|
|
if err != nil {
|
|
return raw
|
|
}
|
|
interfaceInfo, err := ns.siteInfoService.GetSiteInterface(ctx)
|
|
if err != nil {
|
|
return raw
|
|
}
|
|
if len(raw.ReceiverLang) == 0 || raw.ReceiverLang == translator.DefaultLangOption {
|
|
raw.ReceiverLang = interfaceInfo.Language
|
|
}
|
|
raw.QuestionUrl = display.QuestionURL(
|
|
seoInfo.Permalink, siteInfo.SiteUrl,
|
|
msg.NewQuestionTemplateRawData.QuestionID, msg.NewQuestionTemplateRawData.QuestionTitle)
|
|
if len(msg.NewQuestionTemplateRawData.QuestionAuthorUserID) > 0 {
|
|
triggerUser, exist, err := ns.userRepo.GetByUserID(ctx, msg.NewQuestionTemplateRawData.QuestionAuthorUserID)
|
|
if err != nil {
|
|
log.Errorf("get trigger user basic info failed: %v", err)
|
|
return
|
|
}
|
|
if exist {
|
|
raw.TriggerUserID = triggerUser.ID
|
|
raw.TriggerUserDisplayName = triggerUser.DisplayName
|
|
raw.TriggerUserUrl = display.UserURL(siteInfo.SiteUrl, triggerUser.Username)
|
|
}
|
|
}
|
|
return raw
|
|
}
|