381 lines
9.8 KiB
Go
381 lines
9.8 KiB
Go
|
package service
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"context"
|
|||
|
|
"encoding/json"
|
|||
|
|
"fmt"
|
|||
|
|
"time"
|
|||
|
|
|
|||
|
|
"with_you/internal/dto"
|
|||
|
|
"with_you/internal/model"
|
|||
|
|
"with_you/internal/pkg/ws"
|
|||
|
|
"with_you/internal/repository"
|
|||
|
|
|
|||
|
|
"github.com/redis/go-redis/v9"
|
|||
|
|
"go.uber.org/zap"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
// PushWorkerConfig 推送 Worker 配置
|
|||
|
|
type PushWorkerConfig struct {
|
|||
|
|
Enabled bool
|
|||
|
|
Stream string // Redis Stream 名称，默认 "msg_push"
|
|||
|
|
Group string // Consumer Group 名称，默认 "push_worker"
|
|||
|
|
BatchSize int // XREADGROUP 每次读取条数，默认 50
|
|||
|
|
PollTimeoutMs int // XREADGROUP BLOCK 超时 ms，默认 2000
|
|||
|
|
MaxRetries int // 消息最大重试次数，默认 3
|
|||
|
|
MaxStreamLen int64 // Stream MAXLEN，默认 100000
|
|||
|
|
IdleTimeoutMs int64 // XPENDING 空闲超时 ms，默认 30000
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// PushStreamMsg 推送到 Redis Stream 的消息结构
|
|||
|
|
type PushStreamMsg struct {
|
|||
|
|
ConversationID string `json:"conversation_id"`
|
|||
|
|
SenderID string `json:"sender_id"`
|
|||
|
|
MessageID string `json:"message_id"`
|
|||
|
|
ConvType string `json:"conv_type"`
|
|||
|
|
Timestamp int64 `json:"timestamp"`
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// PushWorker 从 Redis Stream 消费消息并异步推送
|
|||
|
|
type PushWorker struct {
|
|||
|
|
rdb *redis.Client
|
|||
|
|
publisher ws.MessagePublisher
|
|||
|
|
pushSvc PushService
|
|||
|
|
msgRepo repository.MessageRepository
|
|||
|
|
userRepo repository.UserRepository
|
|||
|
|
config PushWorkerConfig
|
|||
|
|
stopCh chan struct{}
|
|||
|
|
doneCh chan struct{}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// NewPushWorker 创建推送 Worker
|
|||
|
|
func NewPushWorker(
|
|||
|
|
rdb *redis.Client,
|
|||
|
|
publisher ws.MessagePublisher,
|
|||
|
|
pushSvc PushService,
|
|||
|
|
msgRepo repository.MessageRepository,
|
|||
|
|
userRepo repository.UserRepository,
|
|||
|
|
config PushWorkerConfig,
|
|||
|
|
) *PushWorker {
|
|||
|
|
if config.Stream == "" {
|
|||
|
|
config.Stream = "msg_push"
|
|||
|
|
}
|
|||
|
|
if config.Group == "" {
|
|||
|
|
config.Group = "push_worker"
|
|||
|
|
}
|
|||
|
|
if config.BatchSize <= 0 {
|
|||
|
|
config.BatchSize = 50
|
|||
|
|
}
|
|||
|
|
if config.PollTimeoutMs <= 0 {
|
|||
|
|
config.PollTimeoutMs = 2000
|
|||
|
|
}
|
|||
|
|
if config.MaxRetries <= 0 {
|
|||
|
|
config.MaxRetries = 3
|
|||
|
|
}
|
|||
|
|
if config.MaxStreamLen <= 0 {
|
|||
|
|
config.MaxStreamLen = 100000
|
|||
|
|
}
|
|||
|
|
if config.IdleTimeoutMs <= 0 {
|
|||
|
|
config.IdleTimeoutMs = 30000
|
|||
|
|
}
|
|||
|
|
return &PushWorker{
|
|||
|
|
rdb: rdb,
|
|||
|
|
publisher: publisher,
|
|||
|
|
pushSvc: pushSvc,
|
|||
|
|
msgRepo: msgRepo,
|
|||
|
|
userRepo: userRepo,
|
|||
|
|
config: config,
|
|||
|
|
stopCh: make(chan struct{}),
|
|||
|
|
doneCh: make(chan struct{}),
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// IsEnabled 是否启用
|
|||
|
|
func (w *PushWorker) IsEnabled() bool {
|
|||
|
|
return w != nil && w.config.Enabled
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// Start 启动 Worker
|
|||
|
|
func (w *PushWorker) Start(ctx context.Context) {
|
|||
|
|
if !w.config.Enabled {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 创建 Consumer Group（忽略 BUSYGROUP 错误）
|
|||
|
|
err := w.rdb.XGroupCreateMkStream(ctx, w.config.Stream, w.config.Group, "0").Err()
|
|||
|
|
if err != nil {
|
|||
|
|
// BUSYGROUP 错误可以忽略
|
|||
|
|
zap.L().Debug("XGroupCreateMkStream", zap.Error(err))
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
go w.consumeLoop()
|
|||
|
|
go w.claimPendingLoop()
|
|||
|
|
|
|||
|
|
zap.L().Info("PushWorker started",
|
|||
|
|
zap.String("stream", w.config.Stream),
|
|||
|
|
zap.String("group", w.config.Group),
|
|||
|
|
)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// Stop 停止 Worker
|
|||
|
|
func (w *PushWorker) Stop() {
|
|||
|
|
if !w.config.Enabled {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
close(w.stopCh)
|
|||
|
|
// 等待两个循环退出
|
|||
|
|
<-w.doneCh
|
|||
|
|
<-w.doneCh
|
|||
|
|
zap.L().Info("PushWorker stopped")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// consumeLoop 主消费循环
|
|||
|
|
func (w *PushWorker) consumeLoop() {
|
|||
|
|
defer func() { w.doneCh <- struct{}{} }()
|
|||
|
|
|
|||
|
|
consumerName := fmt.Sprintf("pw-%d", time.Now().UnixNano())
|
|||
|
|
args := &redis.XReadGroupArgs{
|
|||
|
|
Group: w.config.Group,
|
|||
|
|
Consumer: consumerName,
|
|||
|
|
Streams: []string{w.config.Stream, ">"},
|
|||
|
|
Count: int64(w.config.BatchSize),
|
|||
|
|
Block: time.Duration(w.config.PollTimeoutMs) * time.Millisecond,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
for {
|
|||
|
|
select {
|
|||
|
|
case <-w.stopCh:
|
|||
|
|
return
|
|||
|
|
default:
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
results, err := w.rdb.XReadGroup(context.Background(), args).Result()
|
|||
|
|
if err != nil {
|
|||
|
|
if err == redis.Nil {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
zap.L().Warn("PushWorker XReadGroup error", zap.Error(err))
|
|||
|
|
time.Sleep(1 * time.Second)
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
for _, stream := range results {
|
|||
|
|
for _, msg := range stream.Messages {
|
|||
|
|
if err := w.handleMsg(msg); err != nil {
|
|||
|
|
zap.L().Warn("PushWorker handle message error",
|
|||
|
|
zap.String("stream_id", msg.ID),
|
|||
|
|
zap.Error(err),
|
|||
|
|
)
|
|||
|
|
}
|
|||
|
|
// ACK 无论成功失败（失败消息由 claimPendingLoop 重试）
|
|||
|
|
w.rdb.XAck(context.Background(), w.config.Stream, w.config.Group, msg.ID)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// claimPendingLoop 定期认领超时未 ACK 的消息
|
|||
|
|
func (w *PushWorker) claimPendingLoop() {
|
|||
|
|
defer func() { w.doneCh <- struct{}{} }()
|
|||
|
|
|
|||
|
|
consumerName := fmt.Sprintf("pw-claim-%d", time.Now().UnixNano())
|
|||
|
|
ticker := time.NewTicker(30 * time.Second)
|
|||
|
|
defer ticker.Stop()
|
|||
|
|
|
|||
|
|
for {
|
|||
|
|
select {
|
|||
|
|
case <-w.stopCh:
|
|||
|
|
return
|
|||
|
|
case <-ticker.C:
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 查询 pending 消息
|
|||
|
|
pendingResult, err := w.rdb.XPendingExt(context.Background(), &redis.XPendingExtArgs{
|
|||
|
|
Stream: w.config.Stream,
|
|||
|
|
Group: w.config.Group,
|
|||
|
|
Start: "-",
|
|||
|
|
End: "+",
|
|||
|
|
Count: int64(w.config.BatchSize),
|
|||
|
|
Idle: time.Duration(w.config.IdleTimeoutMs) * time.Millisecond,
|
|||
|
|
}).Result()
|
|||
|
|
if err != nil {
|
|||
|
|
if err == redis.Nil {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
zap.L().Warn("PushWorker XPENDING error", zap.Error(err))
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
for _, pending := range pendingResult {
|
|||
|
|
// 认领消息
|
|||
|
|
claimed, err := w.rdb.XClaim(context.Background(), &redis.XClaimArgs{
|
|||
|
|
Stream: w.config.Stream,
|
|||
|
|
Group: w.config.Group,
|
|||
|
|
Consumer: consumerName,
|
|||
|
|
MinIdle: time.Duration(w.config.IdleTimeoutMs) * time.Millisecond,
|
|||
|
|
Messages: []string{pending.ID},
|
|||
|
|
}).Result()
|
|||
|
|
if err != nil || len(claimed) == 0 {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
retryCount := int(pending.RetryCount)
|
|||
|
|
for _, msg := range claimed {
|
|||
|
|
if retryCount >= w.config.MaxRetries {
|
|||
|
|
zap.L().Warn("PushWorker message exceeded max retries, deleting",
|
|||
|
|
zap.String("stream_id", msg.ID),
|
|||
|
|
zap.Int("retry_count", retryCount),
|
|||
|
|
)
|
|||
|
|
w.rdb.XDel(context.Background(), w.config.Stream, msg.ID)
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
if err := w.handleMsg(msg); err != nil {
|
|||
|
|
zap.L().Warn("PushWorker claim retry handle error",
|
|||
|
|
zap.String("stream_id", msg.ID),
|
|||
|
|
zap.Int("retry_count", retryCount),
|
|||
|
|
zap.Error(err),
|
|||
|
|
)
|
|||
|
|
}
|
|||
|
|
w.rdb.XAck(context.Background(), w.config.Stream, w.config.Group, msg.ID)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// handleMsg 处理一条 Stream 消息
|
|||
|
|
func (w *PushWorker) handleMsg(streamMsg redis.XMessage) error {
|
|||
|
|
data, ok := streamMsg.Values["data"]
|
|||
|
|
if !ok {
|
|||
|
|
return fmt.Errorf("missing data field in stream message %s", streamMsg.ID)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
var pushMsg PushStreamMsg
|
|||
|
|
dataStr, ok := data.(string)
|
|||
|
|
if !ok {
|
|||
|
|
return fmt.Errorf("data field is not string in stream message %s", streamMsg.ID)
|
|||
|
|
}
|
|||
|
|
if err := json.Unmarshal([]byte(dataStr), &pushMsg); err != nil {
|
|||
|
|
return fmt.Errorf("unmarshal push message: %w", err)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
ctx := context.Background()
|
|||
|
|
|
|||
|
|
// 加载消息
|
|||
|
|
message, err := w.msgRepo.GetMessageByID(pushMsg.MessageID)
|
|||
|
|
if err != nil {
|
|||
|
|
return fmt.Errorf("load message %s: %w", pushMsg.MessageID, err)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 加载会话
|
|||
|
|
conv, err := w.msgRepo.GetConversation(pushMsg.ConversationID)
|
|||
|
|
if err != nil {
|
|||
|
|
return fmt.Errorf("load conversation %s: %w", pushMsg.ConversationID, err)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 加载参与者
|
|||
|
|
participants, err := w.msgRepo.GetConversationParticipants(pushMsg.ConversationID)
|
|||
|
|
if err != nil {
|
|||
|
|
return fmt.Errorf("load participants for %s: %w", pushMsg.ConversationID, err)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ---- WS Push ----
|
|||
|
|
targetIDs := make([]string, 0, len(participants))
|
|||
|
|
for _, p := range participants {
|
|||
|
|
if conv.Type == model.ConversationTypePrivate && p.UserID == pushMsg.SenderID {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
targetIDs = append(targetIDs, p.UserID)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
detailType := "private"
|
|||
|
|
if conv.Type == model.ConversationTypeGroup {
|
|||
|
|
detailType = "group"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
if len(targetIDs) > 20 && conv.Type == model.ConversationTypeGroup {
|
|||
|
|
w.publisher.PublishToUsersConcurrent(targetIDs, "chat_message", map[string]any{
|
|||
|
|
"detail_type": detailType,
|
|||
|
|
"message": dto.ConvertMessageToResponse(message),
|
|||
|
|
})
|
|||
|
|
} else {
|
|||
|
|
w.publisher.PublishToUsers(targetIDs, "chat_message", map[string]any{
|
|||
|
|
"detail_type": detailType,
|
|||
|
|
"message": dto.ConvertMessageToResponse(message),
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ---- WS unread count push ----
|
|||
|
|
for _, p := range participants {
|
|||
|
|
if p.UserID == pushMsg.SenderID {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
// PushWorker doesn't have conversationCache, skip unread count push here
|
|||
|
|
// The inline push in chat_service still handles this when push worker is disabled
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// ---- JPush for offline users ----
|
|||
|
|
if w.pushSvc == nil {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
sender := ChatMessageSender{ID: pushMsg.SenderID}
|
|||
|
|
if senderUser, sErr := w.userRepo.GetByID(pushMsg.SenderID); sErr == nil {
|
|||
|
|
sender.Name = senderUser.Nickname
|
|||
|
|
sender.Avatar = senderUser.Avatar
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
convName := ""
|
|||
|
|
groupID := ""
|
|||
|
|
groupAvatar := ""
|
|||
|
|
if conv.Type == model.ConversationTypeGroup && conv.Group != nil {
|
|||
|
|
convName = conv.Group.Name
|
|||
|
|
groupID = conv.Group.ID
|
|||
|
|
groupAvatar = conv.Group.Avatar
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
allTargetIDs := make([]string, 0, len(participants))
|
|||
|
|
for _, p := range participants {
|
|||
|
|
if p.UserID == pushMsg.SenderID {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
allTargetIDs = append(allTargetIDs, p.UserID)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
var offlineTargetIDs []string
|
|||
|
|
if w.publisher != nil && len(allTargetIDs) > 0 {
|
|||
|
|
_, offlineTargetIDs = w.publisher.FilterOnline(allTargetIDs)
|
|||
|
|
} else {
|
|||
|
|
offlineTargetIDs = allTargetIDs
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
for _, uid := range offlineTargetIDs {
|
|||
|
|
if pushErr := w.pushSvc.PushChatMessage(ctx, uid, pushMsg.ConversationID, &sender, conv.Type, convName, message, groupID, groupAvatar); pushErr != nil {
|
|||
|
|
zap.L().Debug("push worker: JPush chat message failed",
|
|||
|
|
zap.String("userID", uid),
|
|||
|
|
zap.String("conversationID", pushMsg.ConversationID),
|
|||
|
|
zap.Error(pushErr),
|
|||
|
|
)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// PublishToPush 将消息发布到推送 Stream
|
|||
|
|
func PublishToPush(ctx context.Context, rdb *redis.Client, stream string, msg *PushStreamMsg) error {
|
|||
|
|
data, err := json.Marshal(msg)
|
|||
|
|
if err != nil {
|
|||
|
|
return fmt.Errorf("marshal push message: %w", err)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return rdb.XAdd(ctx, &redis.XAddArgs{
|
|||
|
|
Stream: stream,
|
|||
|
|
MaxLen: 100000,
|
|||
|
|
Approx: true,
|
|||
|
|
ID: "*",
|
|||
|
|
Values: map[string]any{
|
|||
|
|
"data": string(data),
|
|||
|
|
},
|
|||
|
|
}).Err()
|
|||
|
|
}
|