lan
628a6acbe9
Introduce a new mechanism for tracking read positions using message sequences (hasReadSeq), similar to OpenIM, to allow for O(1) unread count calculations. - Implement `UserReadSeq` caching in Redis with Lua scripts to prevent sequence regression (ensuring updates only occur if the new sequence is greater than the current one). - Update `ConversationCache` to support sequence-based unread count computation (`maxSeq - hasReadSeq`). - Refactor `MessageRepository` to use conditional updates for `last_read_seq` in the database. - Update `ChatService` to automatically update the sender's read sequence when sending a message. - Optimize `MarkAsRead` logic to update both database and Redis sequence caches and refine notification targets for private vs group chats. - Update `jpush` client to use pointer types for boolean fields to properly handle optional values in JSON.
1001 lines
33 KiB
Go
1001 lines
33 KiB
Go
package repository
|
||
|
||
import (
|
||
"with_you/internal/model"
|
||
"with_you/internal/pkg/cursor"
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"gorm.io/gorm"
|
||
"gorm.io/gorm/clause"
|
||
)
|
||
|
||
// MessageRepository 消息仓储接口
|
||
type MessageRepository interface {
|
||
CreateMessage(msg *model.Message) error
|
||
GetConversation(id string) (*model.Conversation, error)
|
||
GetOrCreatePrivateConversation(user1ID, user2ID string) (*model.Conversation, error)
|
||
CreateConversationWithParticipants(conv *model.Conversation, participants []string) error
|
||
GetConversations(userID string, page, pageSize int) ([]*model.Conversation, int64, error)
|
||
GetMessages(conversationID string, page, pageSize int) ([]*model.Message, int64, error)
|
||
GetMessagesAfterSeq(conversationID string, afterSeq int64, limit int) ([]*model.Message, error)
|
||
GetMessagesBeforeSeq(conversationID string, beforeSeq int64, limit int) ([]*model.Message, error)
|
||
GetConversationParticipants(conversationID string) ([]*model.ConversationParticipant, error)
|
||
GetParticipant(conversationID string, userID string) (*model.ConversationParticipant, error)
|
||
UpdateLastReadSeq(conversationID string, userID string, lastReadSeq int64) error
|
||
UpdatePinned(conversationID string, userID string, isPinned bool) error
|
||
UpdateNotificationMuted(conversationID string, userID string, notificationMuted bool) error
|
||
GetUnreadCount(conversationID string, userID string) (int64, error)
|
||
UpdateConversationLastSeq(conversationID string, seq int64) error
|
||
GetNextSeq(conversationID string) (int64, error)
|
||
CreateMessageWithSeq(msg *model.Message) error
|
||
GetAllUnreadCount(userID string) (int64, error)
|
||
GetMessageByID(messageID string) (*model.Message, error)
|
||
CountMessagesBySenderInConversation(conversationID, senderID string) (int64, error)
|
||
UpdateMessageStatus(messageID string, status model.MessageStatus) error
|
||
RecallMessage(messageID string, userID string) error
|
||
GetOrCreateSystemParticipant(userID string) (*model.ConversationParticipant, error)
|
||
GetSystemMessagesUnreadCount(userID string) (int64, error)
|
||
MarkAllSystemMessagesAsRead(userID string) error
|
||
GetConversationByGroupID(groupID string) (*model.Conversation, error)
|
||
RemoveParticipant(conversationID string, userID string) error
|
||
AddParticipant(conversationID string, userID string) error
|
||
DeleteConversationByGroupID(groupID string) error
|
||
HideConversationForUser(conversationID, userID string) error
|
||
BatchWriteMessages(ctx context.Context, messages []*model.Message) error
|
||
BatchUpdateParticipants(ctx context.Context, updates []ParticipantUpdate) error
|
||
UpdateConversationLastSeqWithContext(ctx context.Context, convID string, lastSeq int64, lastMsgTime time.Time) error
|
||
BatchWriteMessagesWithTx(tx *gorm.DB, messages []*model.Message) error
|
||
BatchUpdateParticipantsWithTx(tx *gorm.DB, updates []ParticipantUpdate) error
|
||
UpdateConversationLastSeqWithTx(tx *gorm.DB, convID string, lastSeq int64, lastMsgTime time.Time) error
|
||
GetMessagesByCursor(ctx context.Context, conversationID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Message], error)
|
||
GetConversationsByCursor(ctx context.Context, userID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Conversation], error)
|
||
|
||
// 批量查询方法（解决 N+1 问题）
|
||
GetUnreadCountBatch(ctx context.Context, userID string, convIDs []string) (map[string]int64, error)
|
||
GetLastMessagesBatch(ctx context.Context, convIDs []string) (map[string]*model.Message, error)
|
||
GetParticipantsBatch(ctx context.Context, convIDs []string) (map[string][]*model.ConversationParticipant, error)
|
||
GetMyParticipantsBatch(ctx context.Context, convIDs []string, userID string) (map[string]*model.ConversationParticipant, error)
|
||
}
|
||
|
||
// messageRepository 消息仓储实现
|
||
type messageRepository struct {
|
||
db *gorm.DB
|
||
}
|
||
|
||
// NewMessageRepository 创建消息仓储
|
||
func NewMessageRepository(db *gorm.DB) MessageRepository {
|
||
return &messageRepository{db: db}
|
||
}
|
||
|
||
// CreateMessage 创建消息
|
||
func (r *messageRepository) CreateMessage(msg *model.Message) error {
|
||
return r.db.Create(msg).Error
|
||
}
|
||
|
||
// GetConversation 获取会话
|
||
func (r *messageRepository) GetConversation(id string) (*model.Conversation, error) {
|
||
var conv model.Conversation
|
||
err := r.db.Preload("Group").First(&conv, "id = ?", id).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &conv, nil
|
||
}
|
||
|
||
// CreateConversationWithParticipants 创建会话并添加参与者（在事务中）
|
||
func (r *messageRepository) CreateConversationWithParticipants(conv *model.Conversation, participants []string) error {
|
||
return r.db.Transaction(func(tx *gorm.DB) error {
|
||
if err := tx.Create(conv).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
for _, userID := range participants {
|
||
participant := model.ConversationParticipant{
|
||
ConversationID: conv.ID,
|
||
UserID: userID,
|
||
LastReadSeq: 0,
|
||
}
|
||
if err := tx.Create(&participant).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
return nil
|
||
})
|
||
}
|
||
|
||
// GetOrCreatePrivateConversation 获取或创建私聊会话
|
||
// 使用参与者关系表来管理会话
|
||
// userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致
|
||
func (r *messageRepository) GetOrCreatePrivateConversation(user1ID, user2ID string) (*model.Conversation, error) {
|
||
var conv model.Conversation
|
||
|
||
// 查找两个用户共同参与的私聊会话
|
||
err := r.db.Table("conversations c").
|
||
Joins("INNER JOIN conversation_participants cp1 ON c.id = cp1.conversation_id AND cp1.user_id = ?", user1ID).
|
||
Joins("INNER JOIN conversation_participants cp2 ON c.id = cp2.conversation_id AND cp2.user_id = ?", user2ID).
|
||
Where("c.type = ?", model.ConversationTypePrivate).
|
||
First(&conv).Error
|
||
|
||
if err == nil {
|
||
_ = r.db.Model(&model.ConversationParticipant{}).
|
||
Where("conversation_id = ? AND user_id IN ?", conv.ID, []string{user1ID, user2ID}).
|
||
Update("hidden_at", nil).Error
|
||
return &conv, nil
|
||
}
|
||
|
||
if err != gorm.ErrRecordNotFound {
|
||
return nil, err
|
||
}
|
||
|
||
// 没找到会话，创建新会话
|
||
conv = model.Conversation{
|
||
Type: model.ConversationTypePrivate,
|
||
}
|
||
|
||
// 使用事务创建会话和参与者
|
||
err = r.db.Transaction(func(tx *gorm.DB) error {
|
||
if err := tx.Create(&conv).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
// 创建参与者记录 - UserID 存储为 string (UUID)
|
||
participants := []model.ConversationParticipant{
|
||
{ConversationID: conv.ID, UserID: user1ID},
|
||
{ConversationID: conv.ID, UserID: user2ID},
|
||
}
|
||
if err := tx.Create(&participants).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
return nil
|
||
})
|
||
|
||
return &conv, err
|
||
}
|
||
|
||
// GetConversations 获取用户会话列表
|
||
// userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致
|
||
func (r *messageRepository) GetConversations(userID string, page, pageSize int) ([]*model.Conversation, int64, error) {
|
||
var convs []*model.Conversation
|
||
var total int64
|
||
|
||
// 获取总数（私聊仅统计有消息的会话，群聊始终显示）
|
||
r.db.Model(&model.ConversationParticipant{}).
|
||
Joins("INNER JOIN conversations c ON c.id = conversation_participants.conversation_id").
|
||
Where("conversation_participants.user_id = ? AND conversation_participants.hidden_at IS NULL AND (c.type != ? OR c.last_seq > 0)", userID, model.ConversationTypePrivate).
|
||
Count(&total)
|
||
|
||
if total == 0 {
|
||
return convs, total, nil
|
||
}
|
||
|
||
offset := (page - 1) * pageSize
|
||
err := r.db.Model(&model.Conversation{}).
|
||
Joins("INNER JOIN conversation_participants cp ON conversations.id = cp.conversation_id").
|
||
Where("cp.user_id = ? AND cp.hidden_at IS NULL AND (conversations.type != ? OR conversations.last_seq > 0)", userID, model.ConversationTypePrivate).
|
||
Preload("Group").
|
||
Offset(offset).
|
||
Limit(pageSize).
|
||
Order("cp.is_pinned DESC").
|
||
Order("conversations.updated_at DESC").
|
||
Find(&convs).Error
|
||
|
||
return convs, total, err
|
||
}
|
||
|
||
// GetMessages 获取会话消息
|
||
func (r *messageRepository) GetMessages(conversationID string, page, pageSize int) ([]*model.Message, int64, error) {
|
||
var messages []*model.Message
|
||
var total int64
|
||
|
||
r.db.Model(&model.Message{}).Where("conversation_id = ?", conversationID).Count(&total)
|
||
|
||
offset := (page - 1) * pageSize
|
||
err := r.db.Where("conversation_id = ?", conversationID).
|
||
Offset(offset).
|
||
Limit(pageSize).
|
||
Order("seq DESC").
|
||
Find(&messages).Error
|
||
|
||
if err == nil && len(messages) > 0 {
|
||
model.BatchDecryptMessagesParallel(messages)
|
||
}
|
||
|
||
return messages, total, err
|
||
}
|
||
|
||
// GetMessagesAfterSeq 获取指定seq之后的消息（用于增量同步）
|
||
func (r *messageRepository) GetMessagesAfterSeq(conversationID string, afterSeq int64, limit int) ([]*model.Message, error) {
|
||
var messages []*model.Message
|
||
err := r.db.Where("conversation_id = ? AND seq > ?", conversationID, afterSeq).
|
||
Order("seq ASC").
|
||
Limit(limit).
|
||
Find(&messages).Error
|
||
|
||
if err == nil && len(messages) > 0 {
|
||
model.BatchDecryptMessagesParallel(messages)
|
||
}
|
||
|
||
return messages, err
|
||
}
|
||
|
||
// GetMessagesBeforeSeq 获取指定seq之前的历史消息（用于下拉加载更多）
|
||
func (r *messageRepository) GetMessagesBeforeSeq(conversationID string, beforeSeq int64, limit int) ([]*model.Message, error) {
|
||
var messages []*model.Message
|
||
err := r.db.Where("conversation_id = ? AND seq < ?", conversationID, beforeSeq).
|
||
Order("seq DESC"). // 降序获取最新消息在前
|
||
Limit(limit).
|
||
Find(&messages).Error
|
||
// 反转回正序
|
||
for i, j := 0, len(messages)-1; i < j; i, j = i+1, j-1 {
|
||
messages[i], messages[j] = messages[j], messages[i]
|
||
}
|
||
|
||
if err == nil && len(messages) > 0 {
|
||
model.BatchDecryptMessagesParallel(messages)
|
||
}
|
||
|
||
return messages, err
|
||
}
|
||
|
||
// GetConversationParticipants 获取会话参与者
|
||
func (r *messageRepository) GetConversationParticipants(conversationID string) ([]*model.ConversationParticipant, error) {
|
||
var participants []*model.ConversationParticipant
|
||
err := r.db.Where("conversation_id = ?", conversationID).Find(&participants).Error
|
||
return participants, err
|
||
}
|
||
|
||
// GetParticipant 获取用户在会话中的参与者信息
|
||
// userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致
|
||
func (r *messageRepository) GetParticipant(conversationID string, userID string) (*model.ConversationParticipant, error) {
|
||
var participant model.ConversationParticipant
|
||
err := r.db.Where("conversation_id = ? AND user_id = ?", conversationID, userID).First(&participant).Error
|
||
if err != nil {
|
||
// 如果找不到参与者，尝试添加（修复没有参与者记录的问题）
|
||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||
// 检查会话是否存在
|
||
var conv model.Conversation
|
||
if err := r.db.Where("id = ?", conversationID).First(&conv).Error; err == nil {
|
||
// 会话存在，添加参与者
|
||
participant = model.ConversationParticipant{
|
||
ConversationID: conversationID,
|
||
UserID: userID,
|
||
}
|
||
if err := r.db.Create(&participant).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
return &participant, nil
|
||
}
|
||
}
|
||
return nil, err
|
||
}
|
||
return &participant, nil
|
||
}
|
||
|
||
// UpdateLastReadSeq 更新已读位置（防回退：仅当新值大于当前值时才更新）
|
||
// userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致
|
||
func (r *messageRepository) UpdateLastReadSeq(conversationID string, userID string, lastReadSeq int64) error {
|
||
// 使用条件更新防止已读位置回退
|
||
result := r.db.Model(&model.ConversationParticipant{}).
|
||
Where("conversation_id = ? AND user_id = ? AND last_read_seq < ?", conversationID, userID, lastReadSeq).
|
||
Update("last_read_seq", lastReadSeq)
|
||
|
||
if result.Error != nil {
|
||
return result.Error
|
||
}
|
||
|
||
// 如果没有更新任何记录，可能是参与者记录不存在，尝试插入
|
||
if result.RowsAffected == 0 {
|
||
// 检查是否已有记录但 seq 不需要更新
|
||
var existingSeq int64
|
||
err := r.db.Model(&model.ConversationParticipant{}).
|
||
Where("conversation_id = ? AND user_id = ?", conversationID, userID).
|
||
Select("last_read_seq").Scan(&existingSeq).Error
|
||
if err == nil && existingSeq >= lastReadSeq {
|
||
// 已有记录且 seq 不需要回退，直接返回成功
|
||
return nil
|
||
}
|
||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return err
|
||
}
|
||
|
||
// 参与者记录不存在，插入新记录
|
||
err = r.db.Clauses(clause.OnConflict{
|
||
Columns: []clause.Column{
|
||
{Name: "conversation_id"},
|
||
{Name: "user_id"},
|
||
},
|
||
DoUpdates: clause.Assignments(map[string]any{
|
||
"last_read_seq": gorm.Expr("CASE WHEN last_read_seq < ? THEN ? ELSE last_read_seq END", lastReadSeq, lastReadSeq),
|
||
"updated_at": gorm.Expr("CURRENT_TIMESTAMP"),
|
||
}),
|
||
}).Create(&model.ConversationParticipant{
|
||
ConversationID: conversationID,
|
||
UserID: userID,
|
||
LastReadSeq: lastReadSeq,
|
||
}).Error
|
||
if err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// UpdatePinned 更新会话置顶状态（用户维度）
|
||
func (r *messageRepository) UpdatePinned(conversationID string, userID string, isPinned bool) error {
|
||
result := r.db.Model(&model.ConversationParticipant{}).
|
||
Where("conversation_id = ? AND user_id = ?", conversationID, userID).
|
||
Update("is_pinned", isPinned)
|
||
|
||
if result.Error != nil {
|
||
return result.Error
|
||
}
|
||
|
||
if result.RowsAffected == 0 {
|
||
return r.db.Clauses(clause.OnConflict{
|
||
Columns: []clause.Column{
|
||
{Name: "conversation_id"},
|
||
{Name: "user_id"},
|
||
},
|
||
DoUpdates: clause.Assignments(map[string]any{
|
||
"is_pinned": isPinned,
|
||
"updated_at": gorm.Expr("CURRENT_TIMESTAMP"),
|
||
}),
|
||
}).Create(&model.ConversationParticipant{
|
||
ConversationID: conversationID,
|
||
UserID: userID,
|
||
IsPinned: isPinned,
|
||
}).Error
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// UpdateNotificationMuted 更新会话免打扰状态（用户维度）
|
||
func (r *messageRepository) UpdateNotificationMuted(conversationID string, userID string, notificationMuted bool) error {
|
||
result := r.db.Model(&model.ConversationParticipant{}).
|
||
Where("conversation_id = ? AND user_id = ?", conversationID, userID).
|
||
Update("muted", notificationMuted)
|
||
|
||
if result.Error != nil {
|
||
return result.Error
|
||
}
|
||
|
||
if result.RowsAffected == 0 {
|
||
return r.db.Clauses(clause.OnConflict{
|
||
Columns: []clause.Column{
|
||
{Name: "conversation_id"},
|
||
{Name: "user_id"},
|
||
},
|
||
DoUpdates: clause.Assignments(map[string]any{
|
||
"muted": notificationMuted,
|
||
"updated_at": gorm.Expr("CURRENT_TIMESTAMP"),
|
||
}),
|
||
}).Create(&model.ConversationParticipant{
|
||
ConversationID: conversationID,
|
||
UserID: userID,
|
||
NotificationMuted: notificationMuted,
|
||
}).Error
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// GetUnreadCount 获取未读消息数
|
||
// userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致
|
||
func (r *messageRepository) GetUnreadCount(conversationID string, userID string) (int64, error) {
|
||
var participant model.ConversationParticipant
|
||
err := r.db.Where("conversation_id = ? AND user_id = ?", conversationID, userID).First(&participant).Error
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
|
||
var count int64
|
||
err = r.db.Model(&model.Message{}).
|
||
Where("conversation_id = ? AND sender_id != ? AND seq > ?", conversationID, userID, participant.LastReadSeq).
|
||
Count(&count).Error
|
||
return count, err
|
||
}
|
||
|
||
// UpdateConversationLastSeq 更新会话的最后消息seq和时间
|
||
func (r *messageRepository) UpdateConversationLastSeq(conversationID string, seq int64) error {
|
||
return r.db.Model(&model.Conversation{}).
|
||
Where("id = ?", conversationID).
|
||
Updates(map[string]any{
|
||
"last_seq": seq,
|
||
"last_msg_time": gorm.Expr("CURRENT_TIMESTAMP"),
|
||
}).Error
|
||
}
|
||
|
||
// GetNextSeq 获取会话的下一个seq值
|
||
func (r *messageRepository) GetNextSeq(conversationID string) (int64, error) {
|
||
var conv model.Conversation
|
||
err := r.db.Select("last_seq").Where("id = ?", conversationID).First(&conv).Error
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return conv.LastSeq + 1, nil
|
||
}
|
||
|
||
// CreateMessageWithSeq 创建消息并更新seq（事务操作）
|
||
// msg.Seq 应由调用方通过 Redis INCR 预分配
|
||
func (r *messageRepository) CreateMessageWithSeq(msg *model.Message) error {
|
||
return r.db.Transaction(func(tx *gorm.DB) error {
|
||
// 创建消息（seq 已由调用方预分配）
|
||
if err := tx.Create(msg).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
// 更新会话的last_seq
|
||
if err := tx.Model(&model.Conversation{}).
|
||
Where("id = ?", msg.ConversationID).
|
||
Updates(map[string]any{
|
||
"last_seq": msg.Seq,
|
||
"last_msg_time": gorm.Expr("CURRENT_TIMESTAMP"),
|
||
}).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
// 新消息到达后，自动恢复被"仅自己删除"的会话
|
||
if err := tx.Model(&model.ConversationParticipant{}).
|
||
Where("conversation_id = ?", msg.ConversationID).
|
||
Update("hidden_at", nil).Error; err != nil {
|
||
return err
|
||
}
|
||
|
||
return nil
|
||
})
|
||
}
|
||
|
||
// GetAllUnreadCount 获取用户所有会话的未读消息总数
|
||
// userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致
|
||
func (r *messageRepository) GetAllUnreadCount(userID string) (int64, error) {
|
||
var totalUnread int64
|
||
err := r.db.Table("conversation_participants AS cp").
|
||
Joins("LEFT JOIN messages AS m ON m.conversation_id = cp.conversation_id AND m.sender_id <> ? AND m.seq > cp.last_read_seq AND m.deleted_at IS NULL", userID).
|
||
Where("cp.user_id = ?", userID).
|
||
Select("COALESCE(COUNT(m.id), 0)").
|
||
Scan(&totalUnread).Error
|
||
return totalUnread, err
|
||
}
|
||
|
||
// GetMessageByID 根据ID获取消息
|
||
func (r *messageRepository) GetMessageByID(messageID string) (*model.Message, error) {
|
||
var message model.Message
|
||
err := r.db.First(&message, "id = ?", messageID).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// 单条消息直接解密
|
||
_ = message.Decrypt()
|
||
return &message, nil
|
||
}
|
||
|
||
// CountMessagesBySenderInConversation 统计会话中某用户已发送消息数
|
||
func (r *messageRepository) CountMessagesBySenderInConversation(conversationID, senderID string) (int64, error) {
|
||
var count int64
|
||
err := r.db.Model(&model.Message{}).
|
||
Where("conversation_id = ? AND sender_id = ?", conversationID, senderID).
|
||
Count(&count).Error
|
||
return count, err
|
||
}
|
||
|
||
// UpdateMessageStatus 更新消息状态
|
||
func (r *messageRepository) UpdateMessageStatus(messageID string, status model.MessageStatus) error {
|
||
return r.db.Model(&model.Message{}).
|
||
Where("id = ?", messageID).
|
||
Update("status", status).Error
|
||
}
|
||
|
||
// RecallMessage 撤回消息（2分钟内）
|
||
func (r *messageRepository) RecallMessage(messageID string, userID string) error {
|
||
return r.db.Model(&model.Message{}).
|
||
Where("id = ? AND sender_id = ?", messageID, userID).
|
||
Updates(map[string]any{
|
||
"status": model.MessageStatusRecalled,
|
||
"segments": model.MessageSegments{},
|
||
}).Error
|
||
}
|
||
|
||
// GetOrCreateSystemParticipant 获取或创建用户在系统会话中的参与者记录
|
||
// 系统会话是虚拟会话，但需要参与者记录来跟踪已读状态
|
||
func (r *messageRepository) GetOrCreateSystemParticipant(userID string) (*model.ConversationParticipant, error) {
|
||
var participant model.ConversationParticipant
|
||
err := r.db.Where("conversation_id = ? AND user_id = ?",
|
||
model.SystemConversationID, userID).First(&participant).Error
|
||
|
||
if err == nil {
|
||
return &participant, nil
|
||
}
|
||
|
||
if err != gorm.ErrRecordNotFound {
|
||
return nil, err
|
||
}
|
||
|
||
// 自动创建参与者记录
|
||
participant = model.ConversationParticipant{
|
||
ConversationID: model.SystemConversationID,
|
||
UserID: userID,
|
||
LastReadSeq: 0,
|
||
}
|
||
|
||
if err := r.db.Create(&participant).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return &participant, nil
|
||
}
|
||
|
||
// GetSystemMessagesUnreadCount 获取系统消息未读数
|
||
func (r *messageRepository) GetSystemMessagesUnreadCount(userID string) (int64, error) {
|
||
// 获取或创建参与者记录
|
||
participant, err := r.GetOrCreateSystemParticipant(userID)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
|
||
// 计算未读数：查询 seq > last_read_seq 的消息
|
||
var count int64
|
||
err = r.db.Model(&model.Message{}).
|
||
Where("conversation_id = ? AND seq > ?",
|
||
model.SystemConversationID, participant.LastReadSeq).
|
||
Count(&count).Error
|
||
|
||
return count, err
|
||
}
|
||
|
||
// MarkAllSystemMessagesAsRead 标记所有系统消息已读
|
||
func (r *messageRepository) MarkAllSystemMessagesAsRead(userID string) error {
|
||
// 获取系统会话的最新 seq
|
||
var maxSeq int64
|
||
err := r.db.Model(&model.Message{}).
|
||
Where("conversation_id = ?", model.SystemConversationID).
|
||
Select("COALESCE(MAX(seq), 0)").
|
||
Scan(&maxSeq).Error
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 使用跨数据库 upsert 方式更新或创建参与者记录
|
||
return r.db.Clauses(clause.OnConflict{
|
||
Columns: []clause.Column{
|
||
{Name: "conversation_id"},
|
||
{Name: "user_id"},
|
||
},
|
||
DoUpdates: clause.Assignments(map[string]any{
|
||
"last_read_seq": maxSeq,
|
||
"updated_at": gorm.Expr("CURRENT_TIMESTAMP"),
|
||
}),
|
||
}).Create(&model.ConversationParticipant{
|
||
ConversationID: model.SystemConversationID,
|
||
UserID: userID,
|
||
LastReadSeq: maxSeq,
|
||
}).Error
|
||
}
|
||
|
||
// GetConversationByGroupID 通过群组ID获取会话
|
||
func (r *messageRepository) GetConversationByGroupID(groupID string) (*model.Conversation, error) {
|
||
var conv model.Conversation
|
||
err := r.db.Where("group_id = ?", groupID).First(&conv).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &conv, nil
|
||
}
|
||
|
||
// RemoveParticipant 移除会话参与者
|
||
// 当用户退出群聊时，需要同时移除其在对应会话中的参与者记录
|
||
func (r *messageRepository) RemoveParticipant(conversationID string, userID string) error {
|
||
return r.db.Where("conversation_id = ? AND user_id = ?", conversationID, userID).
|
||
Delete(&model.ConversationParticipant{}).Error
|
||
}
|
||
|
||
// AddParticipant 添加会话参与者
|
||
// 当用户加入群聊时，需要同时将其添加到对应会话的参与者记录
|
||
func (r *messageRepository) AddParticipant(conversationID string, userID string) error {
|
||
// 先检查是否已经是参与者
|
||
var count int64
|
||
err := r.db.Model(&model.ConversationParticipant{}).
|
||
Where("conversation_id = ? AND user_id = ?", conversationID, userID).
|
||
Count(&count).Error
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 如果已经是参与者，直接返回
|
||
if count > 0 {
|
||
return nil
|
||
}
|
||
|
||
// 添加参与者
|
||
participant := model.ConversationParticipant{
|
||
ConversationID: conversationID,
|
||
UserID: userID,
|
||
LastReadSeq: 0,
|
||
}
|
||
return r.db.Create(&participant).Error
|
||
}
|
||
|
||
// DeleteConversationByGroupID 删除群组对应的会话及其参与者
|
||
// 当解散群组时调用
|
||
func (r *messageRepository) DeleteConversationByGroupID(groupID string) error {
|
||
// 获取群组对应的会话
|
||
conv, err := r.GetConversationByGroupID(groupID)
|
||
if err != nil {
|
||
// 如果会话不存在，直接返回
|
||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||
return nil
|
||
}
|
||
return err
|
||
}
|
||
|
||
return r.db.Transaction(func(tx *gorm.DB) error {
|
||
// 删除会话参与者
|
||
if err := tx.Where("conversation_id = ?", conv.ID).Delete(&model.ConversationParticipant{}).Error; err != nil {
|
||
return err
|
||
}
|
||
// 删除会话中的消息
|
||
if err := tx.Where("conversation_id = ?", conv.ID).Delete(&model.Message{}).Error; err != nil {
|
||
return err
|
||
}
|
||
// 删除会话
|
||
if err := tx.Delete(&model.Conversation{}, "id = ?", conv.ID).Error; err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
})
|
||
}
|
||
|
||
// HideConversationForUser 仅对当前用户隐藏会话（私聊删除）
|
||
func (r *messageRepository) HideConversationForUser(conversationID, userID string) error {
|
||
now := time.Now()
|
||
return r.db.Model(&model.ConversationParticipant{}).
|
||
Where("conversation_id = ? AND user_id = ?", conversationID, userID).
|
||
Update("hidden_at", &now).Error
|
||
}
|
||
|
||
// ParticipantUpdate 参与者更新数据
|
||
type ParticipantUpdate struct {
|
||
ConversationID string
|
||
UserID string
|
||
LastReadSeq int64
|
||
}
|
||
|
||
// BatchWriteMessages 批量写入消息
|
||
// 使用 GORM 的 CreateInBatches 实现高效批量插入
|
||
func (r *messageRepository) BatchWriteMessages(ctx context.Context, messages []*model.Message) error {
|
||
if len(messages) == 0 {
|
||
return nil
|
||
}
|
||
return r.db.WithContext(ctx).CreateInBatches(messages, 100).Error
|
||
}
|
||
|
||
// BatchUpdateParticipants 批量更新参与者（使用 CASE WHEN 优化）
|
||
// 使用单条 SQL 更新多条记录，避免循环执行 UPDATE
|
||
func (r *messageRepository) BatchUpdateParticipants(ctx context.Context, updates []ParticipantUpdate) error {
|
||
if len(updates) == 0 {
|
||
return nil
|
||
}
|
||
|
||
// 构建 CASE WHEN 批量更新 SQL
|
||
// UPDATE conversation_participants
|
||
// SET last_read_seq = CASE
|
||
// WHEN (conversation_id = '1' AND user_id = 'a') THEN 10
|
||
// WHEN (conversation_id = '2' AND user_id = 'b') THEN 20
|
||
// END,
|
||
// updated_at = ?
|
||
// WHERE (conversation_id = '1' AND user_id = 'a')
|
||
// OR (conversation_id = '2' AND user_id = 'b')
|
||
|
||
var cases []string
|
||
var whereClauses []string
|
||
var args []any
|
||
|
||
for _, u := range updates {
|
||
cases = append(cases, "WHEN (conversation_id = ? AND user_id = ?) THEN ?")
|
||
whereClauses = append(whereClauses, "(conversation_id = ? AND user_id = ?)")
|
||
args = append(args, u.ConversationID, u.UserID, u.LastReadSeq, u.ConversationID, u.UserID)
|
||
}
|
||
|
||
sql := fmt.Sprintf(`
|
||
UPDATE conversation_participants
|
||
SET last_read_seq = CASE %s END,
|
||
updated_at = ?
|
||
WHERE %s
|
||
`, strings.Join(cases, " "), strings.Join(whereClauses, " OR "))
|
||
|
||
args = append(args, time.Now())
|
||
|
||
return r.db.WithContext(ctx).Exec(sql, args...).Error
|
||
}
|
||
|
||
// UpdateConversationLastSeqWithContext 更新会话最后消息序号
|
||
func (r *messageRepository) UpdateConversationLastSeqWithContext(ctx context.Context, convID string, lastSeq int64, lastMsgTime time.Time) error {
|
||
return r.db.WithContext(ctx).
|
||
Model(&model.Conversation{}).
|
||
Where("id = ?", convID).
|
||
Updates(map[string]any{
|
||
"last_seq": lastSeq,
|
||
"last_msg_time": lastMsgTime,
|
||
"updated_at": time.Now(),
|
||
}).Error
|
||
}
|
||
|
||
// BatchWriteMessagesWithTx 在事务中批量写入消息
|
||
func (r *messageRepository) BatchWriteMessagesWithTx(tx *gorm.DB, messages []*model.Message) error {
|
||
if len(messages) == 0 {
|
||
return nil
|
||
}
|
||
return tx.CreateInBatches(messages, 100).Error
|
||
}
|
||
|
||
// BatchUpdateParticipantsWithTx 在事务中批量更新参与者
|
||
func (r *messageRepository) BatchUpdateParticipantsWithTx(tx *gorm.DB, updates []ParticipantUpdate) error {
|
||
if len(updates) == 0 {
|
||
return nil
|
||
}
|
||
|
||
var cases []string
|
||
var whereClauses []string
|
||
var args []any
|
||
|
||
for _, u := range updates {
|
||
cases = append(cases, "WHEN (conversation_id = ? AND user_id = ?) THEN ?")
|
||
whereClauses = append(whereClauses, "(conversation_id = ? AND user_id = ?)")
|
||
args = append(args, u.ConversationID, u.UserID, u.LastReadSeq, u.ConversationID, u.UserID)
|
||
}
|
||
|
||
sql := fmt.Sprintf(`
|
||
UPDATE conversation_participants
|
||
SET last_read_seq = CASE %s END,
|
||
updated_at = ?
|
||
WHERE %s
|
||
`, strings.Join(cases, " "), strings.Join(whereClauses, " OR "))
|
||
|
||
args = append(args, time.Now())
|
||
|
||
return tx.Exec(sql, args...).Error
|
||
}
|
||
|
||
// UpdateConversationLastSeqWithTx 在事务中更新会话最后消息序号
|
||
func (r *messageRepository) UpdateConversationLastSeqWithTx(tx *gorm.DB, convID string, lastSeq int64, lastMsgTime time.Time) error {
|
||
return tx.Model(&model.Conversation{}).
|
||
Where("id = ?", convID).
|
||
Updates(map[string]any{
|
||
"last_seq": lastSeq,
|
||
"last_msg_time": lastMsgTime,
|
||
"updated_at": time.Now(),
|
||
}).Error
|
||
}
|
||
|
||
// ========== Cursor Pagination Methods ==========
|
||
|
||
// GetMessagesByCursor 游标分页获取会话消息
|
||
// 消息按 seq DESC 排序
|
||
func (r *messageRepository) GetMessagesByCursor(ctx context.Context, conversationID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Message], error) {
|
||
// 构建基础查询
|
||
query := r.db.WithContext(ctx).Model(&model.Message{}).Where("conversation_id = ?", conversationID)
|
||
|
||
// 使用游标构建器，消息按 seq DESC 排序
|
||
builder := cursor.NewBuilder(query, cursor.SortBySeqDesc).
|
||
WithCursor(req.Cursor, req.Direction).
|
||
WithPageSize(req.PageSize)
|
||
|
||
if builder.Error() != nil {
|
||
// 无效游标，返回空列表
|
||
return cursor.NewCursorPageResult([]*model.Message{}, "", "", false), nil
|
||
}
|
||
|
||
// 执行查询
|
||
var messages []*model.Message
|
||
if err := builder.Build().Find(&messages).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 解密消息
|
||
if len(messages) > 0 {
|
||
model.BatchDecryptMessagesParallel(messages)
|
||
}
|
||
|
||
// 构建响应
|
||
pageSize := builder.GetPageSize()
|
||
hasMore := cursor.HasMore(len(messages), pageSize)
|
||
if hasMore {
|
||
messages = messages[:pageSize]
|
||
}
|
||
|
||
// 生成游标
|
||
var nextCursor, prevCursor string
|
||
if len(messages) > 0 {
|
||
// 下一页游标
|
||
if hasMore {
|
||
lastMsg := messages[len(messages)-1]
|
||
nextCursor = cursor.NewCursor(
|
||
strconv.FormatInt(lastMsg.Seq, 10),
|
||
lastMsg.ID,
|
||
cursor.SortBySeqDesc,
|
||
).Encode()
|
||
}
|
||
// 上一页游标
|
||
firstMsg := messages[0]
|
||
prevCursor = cursor.NewCursor(
|
||
strconv.FormatInt(firstMsg.Seq, 10),
|
||
firstMsg.ID,
|
||
cursor.SortBySeqDesc,
|
||
).Encode()
|
||
}
|
||
|
||
return cursor.NewCursorPageResult(messages, nextCursor, prevCursor, hasMore), nil
|
||
}
|
||
|
||
// GetConversationsByCursor 游标分页获取用户会话列表
|
||
// 会话按置顶优先，然后按 updated_at DESC 排序
|
||
// 注意：会话列表的排序涉及 conversation_participants 表的 is_pinned 和 updated_at 字段
|
||
func (r *messageRepository) GetConversationsByCursor(ctx context.Context, userID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Conversation], error) {
|
||
// 构建基础查询 - 关联 conversation_participants 表
|
||
query := r.db.WithContext(ctx).Model(&model.Conversation{}).
|
||
Joins("INNER JOIN conversation_participants cp ON conversations.id = cp.conversation_id").
|
||
Where("cp.user_id = ? AND cp.hidden_at IS NULL AND (conversations.type != ? OR conversations.last_seq > 0)", userID, model.ConversationTypePrivate).
|
||
Preload("Group")
|
||
|
||
// 会话列表需要特殊处理：先按置顶排序，再按更新时间排序
|
||
// 由于涉及多字段排序（is_pinned DESC, updated_at DESC），需要自定义处理
|
||
builder := cursor.NewBuilder(query, cursor.SortByUpdatedAtDesc).
|
||
WithCursor(req.Cursor, req.Direction).
|
||
WithPageSize(req.PageSize)
|
||
|
||
if builder.Error() != nil {
|
||
// 无效游标，返回空列表
|
||
return cursor.NewCursorPageResult([]*model.Conversation{}, "", "", false), nil
|
||
}
|
||
|
||
// 执行查询 - 需要添加置顶排序
|
||
var conversations []*model.Conversation
|
||
query = builder.Build()
|
||
// 添加置顶排序（在游标排序之前）
|
||
query = query.Order("cp.is_pinned DESC").Order("conversations.updated_at DESC")
|
||
|
||
if err := query.Find(&conversations).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 构建响应
|
||
pageSize := builder.GetPageSize()
|
||
hasMore := cursor.HasMore(len(conversations), pageSize)
|
||
if hasMore {
|
||
conversations = conversations[:pageSize]
|
||
}
|
||
|
||
// 生成游标
|
||
var nextCursor, prevCursor string
|
||
if len(conversations) > 0 {
|
||
// 下一页游标
|
||
if hasMore {
|
||
lastConv := conversations[len(conversations)-1]
|
||
nextCursor = cursor.NewCursor(
|
||
cursor.FormatTime(lastConv.UpdatedAt),
|
||
lastConv.ID,
|
||
cursor.SortByUpdatedAtDesc,
|
||
).Encode()
|
||
}
|
||
// 上一页游标
|
||
firstConv := conversations[0]
|
||
prevCursor = cursor.NewCursor(
|
||
cursor.FormatTime(firstConv.UpdatedAt),
|
||
firstConv.ID,
|
||
cursor.SortByUpdatedAtDesc,
|
||
).Encode()
|
||
}
|
||
|
||
return cursor.NewCursorPageResult(conversations, nextCursor, prevCursor, hasMore), nil
|
||
}
|
||
|
||
// GetUnreadCountBatch 批量获取用户在多个会话中的未读数
|
||
func (r *messageRepository) GetUnreadCountBatch(ctx context.Context, userID string, convIDs []string) (map[string]int64, error) {
|
||
result := make(map[string]int64, len(convIDs))
|
||
if len(convIDs) == 0 {
|
||
return result, nil
|
||
}
|
||
|
||
type row struct {
|
||
ConversationID string
|
||
Count int64
|
||
}
|
||
var rows []row
|
||
err := r.db.WithContext(ctx).
|
||
Table("messages m").
|
||
Select("m.conversation_id, COUNT(m.id) as count").
|
||
Joins("JOIN conversation_participants cp ON cp.conversation_id = m.conversation_id AND cp.user_id = ?", userID).
|
||
Where("m.sender_id <> ? AND m.deleted_at IS NULL AND m.seq > cp.last_read_seq AND m.conversation_id IN (?)", userID, convIDs).
|
||
Group("m.conversation_id").
|
||
Scan(&rows).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
for _, id := range convIDs {
|
||
result[id] = 0
|
||
}
|
||
for _, r := range rows {
|
||
result[r.ConversationID] = r.Count
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
// GetLastMessagesBatch 批量获取每个会话的最后一条消息
|
||
func (r *messageRepository) GetLastMessagesBatch(ctx context.Context, convIDs []string) (map[string]*model.Message, error) {
|
||
result := make(map[string]*model.Message, len(convIDs))
|
||
if len(convIDs) == 0 {
|
||
return result, nil
|
||
}
|
||
|
||
var messages []*model.Message
|
||
err := r.db.WithContext(ctx).
|
||
Where("id IN (?)", r.db.
|
||
Select("MAX(id)").
|
||
Table("messages").
|
||
Where("conversation_id IN (?) AND deleted_at IS NULL", convIDs).
|
||
Group("conversation_id")).
|
||
Find(&messages).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if len(messages) > 0 {
|
||
model.BatchDecryptMessagesParallel(messages)
|
||
}
|
||
|
||
for _, m := range messages {
|
||
result[m.ConversationID] = m
|
||
}
|
||
return result, nil
|
||
}
|
||
// GetParticipantsBatch 批量获取多个会话的参与者列表
|
||
func (r *messageRepository) GetParticipantsBatch(ctx context.Context, convIDs []string) (map[string][]*model.ConversationParticipant, error) {
|
||
result := make(map[string][]*model.ConversationParticipant, len(convIDs))
|
||
if len(convIDs) == 0 {
|
||
return result, nil
|
||
}
|
||
|
||
var participants []*model.ConversationParticipant
|
||
err := r.db.WithContext(ctx).
|
||
Where("conversation_id IN (?)", convIDs).
|
||
Find(&participants).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
for _, p := range participants {
|
||
result[p.ConversationID] = append(result[p.ConversationID], p)
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
// GetMyParticipantsBatch 批量获取用户在多个会话中的参与者信息
|
||
func (r *messageRepository) GetMyParticipantsBatch(ctx context.Context, convIDs []string, userID string) (map[string]*model.ConversationParticipant, error) {
|
||
result := make(map[string]*model.ConversationParticipant, len(convIDs))
|
||
if len(convIDs) == 0 {
|
||
return result, nil
|
||
}
|
||
|
||
var participants []*model.ConversationParticipant
|
||
err := r.db.WithContext(ctx).
|
||
Where("conversation_id IN (?) AND user_id = ?", convIDs, userID).
|
||
Find(&participants).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
for _, p := range participants {
|
||
result[p.ConversationID] = p
|
||
}
|
||
return result, nil
|
||
}
|