refactor: update repository interfaces and improve dependency injection

- Refactored repository structures to use interfaces instead of concrete types, enhancing flexibility and testability. - Updated various repository methods to accept interfaces, allowing for better dependency management. - Modified wire generation to accommodate new repository interfaces, ensuring proper service injection throughout the application. - Enhanced handler methods to utilize DTOs for filtering and data handling, improving code clarity and maintainability.
2026-03-26 18:14:16 +08:00
parent 7b41dfeb00
commit c6848aba06
50 changed files with 1034 additions and 663 deletions
--- a/internal/repository/message_repo.go
+++ b/internal/repository/message_repo.go
@@ -14,23 +14,64 @@ import (
 	"gorm.io/gorm/clause"
 )

-// MessageRepository 消息仓储
-type MessageRepository struct {
+// 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
+	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)
+}
+
+// messageRepository 消息仓储实现
+type messageRepository struct {
 	db *gorm.DB
 }

 // NewMessageRepository 创建消息仓储
-func NewMessageRepository(db *gorm.DB) *MessageRepository {
-	return &MessageRepository{db: db}
+func NewMessageRepository(db *gorm.DB) MessageRepository {
+	return &messageRepository{db: db}
 }

 // CreateMessage 创建消息
-func (r *MessageRepository) CreateMessage(msg *model.Message) error {
+func (r *messageRepository) CreateMessage(msg *model.Message) error {
 	return r.db.Create(msg).Error
 }

 // GetConversation 获取会话
-func (r *MessageRepository) GetConversation(id string) (*model.Conversation, error) {
+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 {
@@ -39,10 +80,32 @@ func (r *MessageRepository) GetConversation(id string) (*model.Conversation, 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) {
+func (r *messageRepository) GetOrCreatePrivateConversation(user1ID, user2ID string) (*model.Conversation, error) {
 	var conv model.Conversation

 	// 查找两个用户共同参与的私聊会话
@@ -91,7 +154,7 @@ func (r *MessageRepository) GetOrCreatePrivateConversation(user1ID, user2ID stri

 // GetConversations 获取用户会话列表
 // userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致
-func (r *MessageRepository) GetConversations(userID string, page, pageSize int) ([]*model.Conversation, int64, error) {
+func (r *messageRepository) GetConversations(userID string, page, pageSize int) ([]*model.Conversation, int64, error) {
 	var convs []*model.Conversation
 	var total int64

@@ -121,7 +184,7 @@ func (r *MessageRepository) GetConversations(userID string, page, pageSize int)
 }

 // GetMessages 获取会话消息
-func (r *MessageRepository) GetMessages(conversationID string, page, pageSize int) ([]*model.Message, int64, error) {
+func (r *messageRepository) GetMessages(conversationID string, page, pageSize int) ([]*model.Message, int64, error) {
 	var messages []*model.Message
 	var total int64

@@ -142,7 +205,7 @@ func (r *MessageRepository) GetMessages(conversationID string, page, pageSize in
 }

 // GetMessagesAfterSeq 获取指定seq之后的消息（用于增量同步）
-func (r *MessageRepository) GetMessagesAfterSeq(conversationID string, afterSeq int64, limit int) ([]*model.Message, error) {
+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").
@@ -157,7 +220,7 @@ func (r *MessageRepository) GetMessagesAfterSeq(conversationID string, afterSeq
 }

 // GetMessagesBeforeSeq 获取指定seq之前的历史消息（用于下拉加载更多）
-func (r *MessageRepository) GetMessagesBeforeSeq(conversationID string, beforeSeq int64, limit int) ([]*model.Message, error) {
+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"). // 降序获取最新消息在前
@@ -176,7 +239,7 @@ func (r *MessageRepository) GetMessagesBeforeSeq(conversationID string, beforeSe
 }

 // GetConversationParticipants 获取会话参与者
-func (r *MessageRepository) GetConversationParticipants(conversationID string) ([]*model.ConversationParticipant, error) {
+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
@@ -184,7 +247,7 @@ func (r *MessageRepository) GetConversationParticipants(conversationID string) (

 // GetParticipant 获取用户在会话中的参与者信息
 // userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致
-func (r *MessageRepository) GetParticipant(conversationID string, userID string) (*model.ConversationParticipant, error) {
+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 {
@@ -211,7 +274,7 @@ func (r *MessageRepository) GetParticipant(conversationID string, userID string)

 // UpdateLastReadSeq 更新已读位置
 // userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致
-func (r *MessageRepository) UpdateLastReadSeq(conversationID string, userID string, lastReadSeq int64) error {
+func (r *messageRepository) UpdateLastReadSeq(conversationID string, userID string, lastReadSeq int64) error {
 	result := r.db.Model(&model.ConversationParticipant{}).
 		Where("conversation_id = ? AND user_id = ?", conversationID, userID).
 		Update("last_read_seq", lastReadSeq)
@@ -246,7 +309,7 @@ func (r *MessageRepository) UpdateLastReadSeq(conversationID string, userID stri
 }

 // UpdatePinned 更新会话置顶状态（用户维度）
-func (r *MessageRepository) UpdatePinned(conversationID string, userID string, isPinned bool) error {
+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)
@@ -277,7 +340,7 @@ func (r *MessageRepository) UpdatePinned(conversationID string, userID string, i

 // GetUnreadCount 获取未读消息数
 // userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致
-func (r *MessageRepository) GetUnreadCount(conversationID string, userID string) (int64, error) {
+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 {
@@ -292,7 +355,7 @@ func (r *MessageRepository) GetUnreadCount(conversationID string, userID string)
 }

 // UpdateConversationLastSeq 更新会话的最后消息seq和时间
-func (r *MessageRepository) UpdateConversationLastSeq(conversationID string, seq int64) error {
+func (r *messageRepository) UpdateConversationLastSeq(conversationID string, seq int64) error {
 	return r.db.Model(&model.Conversation{}).
 		Where("id = ?", conversationID).
 		Updates(map[string]interface{}{
@@ -302,7 +365,7 @@ func (r *MessageRepository) UpdateConversationLastSeq(conversationID string, seq
 }

 // GetNextSeq 获取会话的下一个seq值
-func (r *MessageRepository) GetNextSeq(conversationID string) (int64, error) {
+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 {
@@ -312,7 +375,7 @@ func (r *MessageRepository) GetNextSeq(conversationID string) (int64, error) {
 }

 // CreateMessageWithSeq 创建消息并更新seq（事务操作）
-func (r *MessageRepository) CreateMessageWithSeq(msg *model.Message) error {
+func (r *messageRepository) CreateMessageWithSeq(msg *model.Message) error {
 	return r.db.Transaction(func(tx *gorm.DB) error {
 		// 获取当前seq并+1
 		var conv model.Conversation
@@ -350,7 +413,7 @@ func (r *MessageRepository) CreateMessageWithSeq(msg *model.Message) error {

 // GetAllUnreadCount 获取用户所有会话的未读消息总数
 // userID 参数为 string 类型（UUID格式），与JWT中user_id保持一致
-func (r *MessageRepository) GetAllUnreadCount(userID string) (int64, error) {
+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).
@@ -361,7 +424,7 @@ func (r *MessageRepository) GetAllUnreadCount(userID string) (int64, error) {
 }

 // GetMessageByID 根据ID获取消息
-func (r *MessageRepository) GetMessageByID(messageID string) (*model.Message, error) {
+func (r *messageRepository) GetMessageByID(messageID string) (*model.Message, error) {
 	var message model.Message
 	err := r.db.First(&message, "id = ?", messageID).Error
 	if err != nil {
@@ -373,7 +436,7 @@ func (r *MessageRepository) GetMessageByID(messageID string) (*model.Message, er
 }

 // CountMessagesBySenderInConversation 统计会话中某用户已发送消息数
-func (r *MessageRepository) CountMessagesBySenderInConversation(conversationID, senderID string) (int64, error) {
+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).
@@ -382,15 +445,25 @@ func (r *MessageRepository) CountMessagesBySenderInConversation(conversationID,
 }

 // UpdateMessageStatus 更新消息状态
-func (r *MessageRepository) UpdateMessageStatus(messageID int64, status model.MessageStatus) error {
+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]interface{}{
+			"status":   model.MessageStatusRecalled,
+			"segments": model.MessageSegments{},
+		}).Error
+}
+
 // GetOrCreateSystemParticipant 获取或创建用户在系统会话中的参与者记录
 // 系统会话是虚拟会话，但需要参与者记录来跟踪已读状态
-func (r *MessageRepository) GetOrCreateSystemParticipant(userID string) (*model.ConversationParticipant, error) {
+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
@@ -418,7 +491,7 @@ func (r *MessageRepository) GetOrCreateSystemParticipant(userID string) (*model.
 }

 // GetSystemMessagesUnreadCount 获取系统消息未读数
-func (r *MessageRepository) GetSystemMessagesUnreadCount(userID string) (int64, error) {
+func (r *messageRepository) GetSystemMessagesUnreadCount(userID string) (int64, error) {
 	// 获取或创建参与者记录
 	participant, err := r.GetOrCreateSystemParticipant(userID)
 	if err != nil {
@@ -436,7 +509,7 @@ func (r *MessageRepository) GetSystemMessagesUnreadCount(userID string) (int64,
 }

 // MarkAllSystemMessagesAsRead 标记所有系统消息已读
-func (r *MessageRepository) MarkAllSystemMessagesAsRead(userID string) error {
+func (r *messageRepository) MarkAllSystemMessagesAsRead(userID string) error {
 	// 获取系统会话的最新 seq
 	var maxSeq int64
 	err := r.db.Model(&model.Message{}).
@@ -465,7 +538,7 @@ func (r *MessageRepository) MarkAllSystemMessagesAsRead(userID string) error {
 }

 // GetConversationByGroupID 通过群组ID获取会话
-func (r *MessageRepository) GetConversationByGroupID(groupID string) (*model.Conversation, error) {
+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 {
@@ -476,14 +549,14 @@ func (r *MessageRepository) GetConversationByGroupID(groupID string) (*model.Con

 // RemoveParticipant 移除会话参与者
 // 当用户退出群聊时，需要同时移除其在对应会话中的参与者记录
-func (r *MessageRepository) RemoveParticipant(conversationID string, userID string) error {
+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 {
+func (r *messageRepository) AddParticipant(conversationID string, userID string) error {
 	// 先检查是否已经是参与者
 	var count int64
 	err := r.db.Model(&model.ConversationParticipant{}).
@@ -509,7 +582,7 @@ func (r *MessageRepository) AddParticipant(conversationID string, userID string)

 // DeleteConversationByGroupID 删除群组对应的会话及其参与者
 // 当解散群组时调用
-func (r *MessageRepository) DeleteConversationByGroupID(groupID string) error {
+func (r *messageRepository) DeleteConversationByGroupID(groupID string) error {
 	// 获取群组对应的会话
 	conv, err := r.GetConversationByGroupID(groupID)
 	if err != nil {
@@ -538,7 +611,7 @@ func (r *MessageRepository) DeleteConversationByGroupID(groupID string) error {
 }

 // HideConversationForUser 仅对当前用户隐藏会话（私聊删除）
-func (r *MessageRepository) HideConversationForUser(conversationID, userID string) error {
+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).
@@ -554,7 +627,7 @@ type ParticipantUpdate struct {

 // BatchWriteMessages 批量写入消息
 // 使用 GORM 的 CreateInBatches 实现高效批量插入
-func (r *MessageRepository) BatchWriteMessages(ctx context.Context, messages []*model.Message) error {
+func (r *messageRepository) BatchWriteMessages(ctx context.Context, messages []*model.Message) error {
 	if len(messages) == 0 {
 		return nil
 	}
@@ -563,7 +636,7 @@ func (r *MessageRepository) BatchWriteMessages(ctx context.Context, messages []*

 // BatchUpdateParticipants 批量更新参与者（使用 CASE WHEN 优化）
 // 使用单条 SQL 更新多条记录，避免循环执行 UPDATE
-func (r *MessageRepository) BatchUpdateParticipants(ctx context.Context, updates []ParticipantUpdate) error {
+func (r *messageRepository) BatchUpdateParticipants(ctx context.Context, updates []ParticipantUpdate) error {
 	if len(updates) == 0 {
 		return nil
 	}
@@ -601,7 +674,7 @@ func (r *MessageRepository) BatchUpdateParticipants(ctx context.Context, updates
 }

 // UpdateConversationLastSeqWithContext 更新会话最后消息序号
-func (r *MessageRepository) UpdateConversationLastSeqWithContext(ctx context.Context, convID string, lastSeq int64, lastMsgTime time.Time) error {
+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).
@@ -613,7 +686,7 @@ func (r *MessageRepository) UpdateConversationLastSeqWithContext(ctx context.Con
 }

 // BatchWriteMessagesWithTx 在事务中批量写入消息
-func (r *MessageRepository) BatchWriteMessagesWithTx(tx *gorm.DB, messages []*model.Message) error {
+func (r *messageRepository) BatchWriteMessagesWithTx(tx *gorm.DB, messages []*model.Message) error {
 	if len(messages) == 0 {
 		return nil
 	}
@@ -621,7 +694,7 @@ func (r *MessageRepository) BatchWriteMessagesWithTx(tx *gorm.DB, messages []*mo
 }

 // BatchUpdateParticipantsWithTx 在事务中批量更新参与者
-func (r *MessageRepository) BatchUpdateParticipantsWithTx(tx *gorm.DB, updates []ParticipantUpdate) error {
+func (r *messageRepository) BatchUpdateParticipantsWithTx(tx *gorm.DB, updates []ParticipantUpdate) error {
 	if len(updates) == 0 {
 		return nil
 	}
@@ -649,7 +722,7 @@ func (r *MessageRepository) BatchUpdateParticipantsWithTx(tx *gorm.DB, updates [
 }

 // UpdateConversationLastSeqWithTx 在事务中更新会话最后消息序号
-func (r *MessageRepository) UpdateConversationLastSeqWithTx(tx *gorm.DB, convID string, lastSeq int64, lastMsgTime time.Time) error {
+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]interface{}{
@@ -663,7 +736,7 @@ func (r *MessageRepository) UpdateConversationLastSeqWithTx(tx *gorm.DB, convID

 // GetMessagesByCursor 游标分页获取会话消息
 // 消息按 seq DESC 排序
-func (r *MessageRepository) GetMessagesByCursor(ctx context.Context, conversationID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Message], error) {
+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)

@@ -722,7 +795,7 @@ func (r *MessageRepository) GetMessagesByCursor(ctx context.Context, conversatio
 // 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) {
+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").