From e091fbccb343fdebd747a84482216b9289d018ed Mon Sep 17 00:00:00 2001 From: bujinyuan Date: Tue, 3 Mar 2026 10:44:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(im):=20Task=202=20-=20IM=20DAO=20=E5=B1=82?= =?UTF-8?q?=EF=BC=88=E4=BC=9A=E8=AF=9D=20+=20=E6=B6=88=E6=81=AF=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ConversationDAO: - FindPrivateConversation: 单聊会话去重查询 - CreateWithMembers: 事务创建会话+成员 - GetUserConversations: 会话列表（含未读/置顶） - UpdateLastMessage / IncrementUnread / ClearUnread - SoftDeleteMember / RestoreMember / UpdateMemberPinned - GetUnreadConversations: 离线消息推送用 MessageDAO: - Create / GetByID / GetByConversation（游标分页） - UpdateStatus（撤回） / DeleteByConversation（清空） - SearchMessages（全局搜索，JOIN 权限校验） - FindByClientMsgID（幂等去重） Made-with: Cursor --- .../go-service/app/im/dao/conversation_dao.go | 255 ++++++++++++++++++ backend/go-service/app/im/dao/message_dao.go | 157 +++++++++++ 2 files changed, 412 insertions(+) create mode 100644 backend/go-service/app/im/dao/conversation_dao.go create mode 100644 backend/go-service/app/im/dao/message_dao.go diff --git a/backend/go-service/app/im/dao/conversation_dao.go b/backend/go-service/app/im/dao/conversation_dao.go new file mode 100644 index 0000000..7b8cd56 --- /dev/null +++ b/backend/go-service/app/im/dao/conversation_dao.go @@ -0,0 +1,255 @@ +// Package dao 提供 IM 模块的数据库访问操作 +package dao + +import ( + "context" + "time" + + "github.com/echochat/backend/app/constants" + "github.com/echochat/backend/app/im/model" + "github.com/echochat/backend/pkg/logs" + "go.uber.org/zap" + "gorm.io/gorm" +) + +// ConversationDAO 会话数据访问对象 +type ConversationDAO struct { + db *gorm.DB +} + +// NewConversationDAO 创建 ConversationDAO 实例 +func NewConversationDAO(db *gorm.DB) *ConversationDAO { + return &ConversationDAO{db: db} +} + +// FindPrivateConversation 查找两人之间的单聊会话 +// 通过 JOIN im_conversation_members 判断是否已存在，避免重复创建 +func (d *ConversationDAO) FindPrivateConversation(ctx context.Context, userID, targetUserID int64) (*model.Conversation, error) { + funcName := "dao.conversation_dao.FindPrivateConversation" + logs.Debug(ctx, funcName, "查找单聊会话", + zap.Int64("user_id", userID), zap.Int64("target_user_id", targetUserID)) + + var conv model.Conversation + err := d.db.WithContext(ctx). + Raw(`SELECT c.* FROM im_conversations c + JOIN im_conversation_members cm1 ON cm1.conversation_id = c.id + JOIN im_conversation_members cm2 ON cm2.conversation_id = c.id + WHERE cm1.user_id = ? AND cm2.user_id = ? AND c.type = ? + LIMIT 1`, + userID, targetUserID, constants.ConversationTypePrivate). + Scan(&conv).Error + + if err != nil { + logs.Error(ctx, funcName, "查找单聊会话失败", zap.Error(err)) + return nil, err + } + if conv.ID == 0 { + return nil, nil + } + return &conv, nil +} + +// CreateWithMembers 在事务中创建会话及其成员记录 +func (d *ConversationDAO) CreateWithMembers(ctx context.Context, conv *model.Conversation, memberIDs []int64) error { + funcName := "dao.conversation_dao.CreateWithMembers" + logs.Info(ctx, funcName, "创建会话及成员", + zap.Int("type", conv.Type), zap.Int("member_count", len(memberIDs))) + + return d.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Create(conv).Error; err != nil { + logs.Error(ctx, funcName, "创建会话失败", zap.Error(err)) + return err + } + + for _, uid := range memberIDs { + member := &model.ConversationMember{ + ConversationID: conv.ID, + UserID: uid, + } + if err := tx.Create(member).Error; err != nil { + logs.Error(ctx, funcName, "创建会话成员失败", + zap.Int64("conversation_id", conv.ID), zap.Int64("user_id", uid), zap.Error(err)) + return err + } + } + return nil + }) +} + +// GetUserConversations 获取用户的会话列表（排除已软删除的） +// 返回会话基本信息和该用户的成员视图（未读数、置顶等） +func (d *ConversationDAO) GetUserConversations(ctx context.Context, userID int64) ([]ConversationWithMember, error) { + funcName := "dao.conversation_dao.GetUserConversations" + logs.Debug(ctx, funcName, "查询用户会话列表", zap.Int64("user_id", userID)) + + var results []ConversationWithMember + err := d.db.WithContext(ctx). + Raw(`SELECT c.id, c.type, c.last_msg_content, c.last_msg_time, c.last_msg_sender_id, + cm.is_pinned, cm.unread_count + FROM im_conversations c + JOIN im_conversation_members cm ON cm.conversation_id = c.id + WHERE cm.user_id = ? AND cm.is_deleted = false + ORDER BY cm.is_pinned DESC, c.last_msg_time DESC NULLS LAST`, + userID). + Scan(&results).Error + + if err != nil { + logs.Error(ctx, funcName, "查询用户会话列表失败", zap.Error(err)) + } + return results, err +} + +// ConversationWithMember 会话列表查询结果（JOIN 成员表） +type ConversationWithMember struct { + ID int64 `json:"id"` + Type int `json:"type"` + LastMsgContent string `json:"last_msg_content"` + LastMsgTime *time.Time `json:"last_msg_time"` + LastMsgSenderID *int64 `json:"last_msg_sender_id"` + IsPinned bool `json:"is_pinned"` + UnreadCount int `json:"unread_count"` +} + +// GetMember 获取指定会话中指定用户的成员记录 +func (d *ConversationDAO) GetMember(ctx context.Context, conversationID, userID int64) (*model.ConversationMember, error) { + funcName := "dao.conversation_dao.GetMember" + logs.Debug(ctx, funcName, "查询会话成员", + zap.Int64("conversation_id", conversationID), zap.Int64("user_id", userID)) + + var member model.ConversationMember + err := d.db.WithContext(ctx). + Where("conversation_id = ? AND user_id = ?", conversationID, userID). + First(&member).Error + if err != nil { + return nil, err + } + return &member, nil +} + +// GetPeerUserID 获取单聊会话中对方的用户 ID +func (d *ConversationDAO) GetPeerUserID(ctx context.Context, conversationID, userID int64) (int64, error) { + funcName := "dao.conversation_dao.GetPeerUserID" + logs.Debug(ctx, funcName, "查询对方用户 ID", + zap.Int64("conversation_id", conversationID), zap.Int64("user_id", userID)) + + var peerID int64 + err := d.db.WithContext(ctx). + Model(&model.ConversationMember{}). + Select("user_id"). + Where("conversation_id = ? AND user_id != ?", conversationID, userID). + Scan(&peerID).Error + return peerID, err +} + +// GetConversationMemberIDs 获取会话的所有成员 ID +func (d *ConversationDAO) GetConversationMemberIDs(ctx context.Context, conversationID int64) ([]int64, error) { + var ids []int64 + err := d.db.WithContext(ctx). + Model(&model.ConversationMember{}). + Where("conversation_id = ?", conversationID). + Pluck("user_id", &ids).Error + return ids, err +} + +// UpdateMemberPinned 更新会话成员的置顶状态 +func (d *ConversationDAO) UpdateMemberPinned(ctx context.Context, conversationID, userID int64, isPinned bool) error { + funcName := "dao.conversation_dao.UpdateMemberPinned" + logs.Info(ctx, funcName, "更新置顶状态", + zap.Int64("conversation_id", conversationID), zap.Int64("user_id", userID), zap.Bool("is_pinned", isPinned)) + + return d.db.WithContext(ctx). + Model(&model.ConversationMember{}). + Where("conversation_id = ? AND user_id = ?", conversationID, userID). + Update("is_pinned", isPinned).Error +} + +// SoftDeleteMember 软删除会话（仅影响当前用户视图，不影响对方） +func (d *ConversationDAO) SoftDeleteMember(ctx context.Context, conversationID, userID int64) error { + funcName := "dao.conversation_dao.SoftDeleteMember" + logs.Info(ctx, funcName, "软删除会话", + zap.Int64("conversation_id", conversationID), zap.Int64("user_id", userID)) + + return d.db.WithContext(ctx). + Model(&model.ConversationMember{}). + Where("conversation_id = ? AND user_id = ?", conversationID, userID). + Updates(map[string]interface{}{ + "is_deleted": true, + "unread_count": 0, + }).Error +} + +// RestoreMember 恢复已软删除的会话成员（收到新消息时自动恢复） +func (d *ConversationDAO) RestoreMember(ctx context.Context, conversationID, userID int64) error { + return d.db.WithContext(ctx). + Model(&model.ConversationMember{}). + Where("conversation_id = ? AND user_id = ?", conversationID, userID). + Update("is_deleted", false).Error +} + +// UpdateLastMessage 更新会话的最后消息信息（冗余字段，避免列表查询 JOIN） +func (d *ConversationDAO) UpdateLastMessage(ctx context.Context, conversationID int64, msgID int64, content string, senderID int64, msgTime time.Time) error { + funcName := "dao.conversation_dao.UpdateLastMessage" + logs.Debug(ctx, funcName, "更新最后消息", + zap.Int64("conversation_id", conversationID), zap.Int64("message_id", msgID)) + + return d.db.WithContext(ctx). + Model(&model.Conversation{}). + Where("id = ?", conversationID). + Updates(map[string]interface{}{ + "last_message_id": msgID, + "last_msg_content": content, + "last_msg_time": msgTime, + "last_msg_sender_id": senderID, + }).Error +} + +// IncrementUnread 将指定成员的未读消息数 +1 +func (d *ConversationDAO) IncrementUnread(ctx context.Context, conversationID, userID int64) error { + return d.db.WithContext(ctx). + Model(&model.ConversationMember{}). + Where("conversation_id = ? AND user_id = ?", conversationID, userID). + UpdateColumn("unread_count", gorm.Expr("unread_count + 1")).Error +} + +// ClearUnread 清零指定成员的未读消息数 +func (d *ConversationDAO) ClearUnread(ctx context.Context, conversationID, userID int64, lastMsgID int64) error { + return d.db.WithContext(ctx). + Model(&model.ConversationMember{}). + Where("conversation_id = ? AND user_id = ?", conversationID, userID). + Updates(map[string]interface{}{ + "unread_count": 0, + "last_read_msg_id": lastMsgID, + }).Error +} + +// GetUnreadConversations 获取有未读消息的会话列表（用于离线消息推送） +func (d *ConversationDAO) GetUnreadConversations(ctx context.Context, userID int64) ([]ConversationWithMember, error) { + funcName := "dao.conversation_dao.GetUnreadConversations" + logs.Debug(ctx, funcName, "查询未读会话", zap.Int64("user_id", userID)) + + var results []ConversationWithMember + err := d.db.WithContext(ctx). + Raw(`SELECT c.id, c.type, c.last_msg_content, c.last_msg_time, c.last_msg_sender_id, + cm.is_pinned, cm.unread_count + FROM im_conversations c + JOIN im_conversation_members cm ON cm.conversation_id = c.id + WHERE cm.user_id = ? AND cm.is_deleted = false AND cm.unread_count > 0 + ORDER BY c.last_msg_time DESC`, + userID). + Scan(&results).Error + + if err != nil { + logs.Error(ctx, funcName, "查询未读会话失败", zap.Error(err)) + } + return results, err +} + +// GetByID 根据 ID 获取会话 +func (d *ConversationDAO) GetByID(ctx context.Context, id int64) (*model.Conversation, error) { + var conv model.Conversation + err := d.db.WithContext(ctx).First(&conv, id).Error + if err != nil { + return nil, err + } + return &conv, nil +} diff --git a/backend/go-service/app/im/dao/message_dao.go b/backend/go-service/app/im/dao/message_dao.go new file mode 100644 index 0000000..8ef04de --- /dev/null +++ b/backend/go-service/app/im/dao/message_dao.go @@ -0,0 +1,157 @@ +package dao + +import ( + "context" + + "github.com/echochat/backend/app/constants" + "github.com/echochat/backend/app/im/model" + "github.com/echochat/backend/pkg/logs" + "go.uber.org/zap" + "gorm.io/gorm" +) + +// MessageDAO 消息数据访问对象 +type MessageDAO struct { + db *gorm.DB +} + +// NewMessageDAO 创建 MessageDAO 实例 +func NewMessageDAO(db *gorm.DB) *MessageDAO { + return &MessageDAO{db: db} +} + +// Create 写入一条消息 +func (d *MessageDAO) Create(ctx context.Context, msg *model.Message) error { + funcName := "dao.message_dao.Create" + logs.Debug(ctx, funcName, "写入消息", + zap.Int64("conversation_id", msg.ConversationID), zap.Int64("sender_id", msg.SenderID)) + + err := d.db.WithContext(ctx).Create(msg).Error + if err != nil { + logs.Error(ctx, funcName, "写入消息失败", zap.Error(err)) + } + return err +} + +// GetByID 根据 ID 查询单条消息 +func (d *MessageDAO) GetByID(ctx context.Context, id int64) (*model.Message, error) { + var msg model.Message + err := d.db.WithContext(ctx).First(&msg, id).Error + if err != nil { + return nil, err + } + return &msg, nil +} + +// GetByConversation 获取会话的历史消息（游标分页，按 ID 降序） +// beforeID > 0 时作为游标，查询 ID 小于 beforeID 的消息 +// beforeID = 0 时查询最新的 limit 条消息 +func (d *MessageDAO) GetByConversation(ctx context.Context, conversationID int64, beforeID int64, limit int) ([]model.Message, error) { + funcName := "dao.message_dao.GetByConversation" + logs.Debug(ctx, funcName, "查询历史消息", + zap.Int64("conversation_id", conversationID), + zap.Int64("before_id", beforeID), zap.Int("limit", limit)) + + query := d.db.WithContext(ctx). + Where("conversation_id = ? AND status != ?", conversationID, constants.MessageStatusDeleted) + + if beforeID > 0 { + query = query.Where("id < ?", beforeID) + } + + var messages []model.Message + err := query.Order("id DESC").Limit(limit).Find(&messages).Error + if err != nil { + logs.Error(ctx, funcName, "查询历史消息失败", zap.Error(err)) + } + return messages, err +} + +// UpdateStatus 更新消息状态（撤回/删除） +func (d *MessageDAO) UpdateStatus(ctx context.Context, id int64, status int) error { + funcName := "dao.message_dao.UpdateStatus" + logs.Info(ctx, funcName, "更新消息状态", + zap.Int64("message_id", id), zap.Int("status", status)) + + return d.db.WithContext(ctx). + Model(&model.Message{}). + Where("id = ?", id). + Update("status", status).Error +} + +// DeleteByConversation 软删除会话中所有消息（标记 status=3） +func (d *MessageDAO) DeleteByConversation(ctx context.Context, conversationID int64) error { + funcName := "dao.message_dao.DeleteByConversation" + logs.Info(ctx, funcName, "清空会话消息", + zap.Int64("conversation_id", conversationID)) + + return d.db.WithContext(ctx). + Model(&model.Message{}). + Where("conversation_id = ? AND status = ?", conversationID, constants.MessageStatusNormal). + Update("status", constants.MessageStatusDeleted).Error +} + +// SearchMessages 全局消息搜索（按关键词匹配，仅搜索用户所在会话的消息） +// 返回匹配的消息列表（已 JOIN 成员表确保权限） +func (d *MessageDAO) SearchMessages(ctx context.Context, userID int64, keyword string, limit int) ([]MessageSearchResult, error) { + funcName := "dao.message_dao.SearchMessages" + logs.Debug(ctx, funcName, "全局消息搜索", + zap.Int64("user_id", userID), zap.String("keyword", keyword)) + + var results []MessageSearchResult + err := d.db.WithContext(ctx). + Raw(`SELECT m.id, m.conversation_id, m.sender_id, m.content, m.created_at + FROM im_messages m + JOIN im_conversation_members cm ON cm.conversation_id = m.conversation_id + WHERE cm.user_id = ? AND cm.is_deleted = false + AND m.status = ? AND m.content LIKE ? + ORDER BY m.created_at DESC + LIMIT ?`, + userID, constants.MessageStatusNormal, "%"+keyword+"%", limit). + Scan(&results).Error + + if err != nil { + logs.Error(ctx, funcName, "全局消息搜索失败", zap.Error(err)) + } + return results, err +} + +// MessageSearchResult 搜索结果单条记录 +type MessageSearchResult struct { + ID int64 `json:"id"` + ConversationID int64 `json:"conversation_id"` + SenderID int64 `json:"sender_id"` + Content string `json:"content"` + CreatedAt string `json:"created_at"` +} + +// FindByClientMsgID 根据客户端消息 ID 查询（幂等去重用） +func (d *MessageDAO) FindByClientMsgID(ctx context.Context, conversationID int64, clientMsgID string) (*model.Message, error) { + if clientMsgID == "" { + return nil, nil + } + + var msg model.Message + err := d.db.WithContext(ctx). + Where("conversation_id = ? AND client_msg_id = ?", conversationID, clientMsgID). + First(&msg).Error + + if err == gorm.ErrRecordNotFound { + return nil, nil + } + if err != nil { + return nil, err + } + return &msg, nil +} + +// GetLatestMessageID 获取会话中最新一条消息的 ID（用于标记已读） +func (d *MessageDAO) GetLatestMessageID(ctx context.Context, conversationID int64) (int64, error) { + var id int64 + err := d.db.WithContext(ctx). + Model(&model.Message{}). + Select("COALESCE(MAX(id), 0)"). + Where("conversation_id = ?", conversationID). + Scan(&id).Error + return id, err +}