feat(im): Task 2 - IM DAO 层（会话 + 消息）

ConversationDAO:
- FindPrivateConversation: 单聊会话去重查询
- CreateWithMembers: 事务创建会话+成员
- GetUserConversations: 会话列表（含未读/置顶）
- UpdateLastMessage / IncrementUnread / ClearUnread
- SoftDeleteMember / RestoreMember / UpdateMemberPinned
- GetUnreadConversations: 离线消息推送用

MessageDAO:
- Create / GetByID / GetByConversation（游标分页）
- UpdateStatus（撤回） / DeleteByConversation（清空）
- SearchMessages（全局搜索，JOIN 权限校验）
- FindByClientMsgID（幂等去重）

Made-with: Cursor

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
+}

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
+}