Files
ai-agent/internal/services/message_service.go
T

671 lines
24 KiB
Go

package services
import (
"cs-agent/internal/models"
"cs-agent/internal/pkg/dto"
"cs-agent/internal/pkg/enums"
"cs-agent/internal/pkg/errorsx"
"cs-agent/internal/pkg/openidentity"
"cs-agent/internal/pkg/utils"
"cs-agent/internal/repositories"
"log/slog"
"slices"
"strings"
"time"
"github.com/mlogclub/simple/common/strs"
"github.com/mlogclub/simple/sqls"
"github.com/mlogclub/simple/web/params"
)
var MessageService = newMessageService()
func newMessageService() *messageService {
return &messageService{}
}
type messageService struct {
}
func (s *messageService) Get(id int64) *models.Message {
return repositories.MessageRepository.Get(sqls.DB(), id)
}
func (s *messageService) Take(where ...interface{}) *models.Message {
return repositories.MessageRepository.Take(sqls.DB(), where...)
}
func (s *messageService) Find(cnd *sqls.Cnd) []models.Message {
return repositories.MessageRepository.Find(sqls.DB(), cnd)
}
// FindByConversationIDCursor 按 id 游标分页:cursor=0 取最新 limit 条;cursor>0 取 id<cursor 的更旧消息。
// 返回的 list 已按 id 升序(时间正序)。nextCursor 为下一页请求传入的游标(本批最小 id);hasMore 表示可能还有更旧消息。
func (s *messageService) FindByConversationIDCursor(conversationID int64, cursor int64, limit int, senderType, messageType string) (list []models.Message, nextCursor int64, hasMore bool) {
if limit > 100 {
limit = 100
} else if limit <= 0 {
limit = 20
}
cnd := sqls.NewCnd().Eq("conversation_id", conversationID).Limit(limit).Desc("id")
if cursor > 0 {
cnd.Lt("id", cursor)
}
if strs.IsNotBlank(senderType) {
cnd.Eq("sender_type", senderType)
}
if strs.IsNotBlank(messageType) {
cnd.Eq("message_type", messageType)
}
list = s.Find(cnd)
nextCursor = cursor
hasMore = false
if len(list) > 0 {
nextCursor = list[len(list)-1].ID
hasMore = len(list) == limit
}
slices.Reverse(list)
return list, nextCursor, hasMore
}
func (s *messageService) FindOne(cnd *sqls.Cnd) *models.Message {
return repositories.MessageRepository.FindOne(sqls.DB(), cnd)
}
func (s *messageService) FindPageByParams(params *params.QueryParams) (list []models.Message, paging *sqls.Paging) {
return repositories.MessageRepository.FindPageByParams(sqls.DB(), params)
}
func (s *messageService) FindPageByCnd(cnd *sqls.Cnd) (list []models.Message, paging *sqls.Paging) {
return repositories.MessageRepository.FindPageByCnd(sqls.DB(), cnd)
}
// FindPageByCndForImListAscending 与 FindPageByCnd 相同分页条件,将结果按 seq 升序排列(开放 IM 时间正序展示)。
func (s *messageService) FindPageByCndForImListAscending(cnd *sqls.Cnd) (list []models.Message, paging *sqls.Paging) {
list, paging = s.FindPageByCnd(cnd)
if len(list) <= 1 {
return list, paging
}
for i, j := 0, len(list)-1; i < j; i, j = i+1, j-1 {
list[i], list[j] = list[j], list[i]
}
return list, paging
}
func (s *messageService) Count(cnd *sqls.Cnd) int64 {
return repositories.MessageRepository.Count(sqls.DB(), cnd)
}
func (s *messageService) Create(t *models.Message) error {
return repositories.MessageRepository.Create(sqls.DB(), t)
}
func (s *messageService) Update(t *models.Message) error {
return repositories.MessageRepository.Update(sqls.DB(), t)
}
func (s *messageService) Updates(id int64, columns map[string]interface{}) error {
return repositories.MessageRepository.Updates(sqls.DB(), id, columns)
}
func (s *messageService) UpdateColumn(id int64, name string, value interface{}) error {
return repositories.MessageRepository.UpdateColumn(sqls.DB(), id, name, value)
}
func (s *messageService) Delete(id int64) {
repositories.MessageRepository.Delete(sqls.DB(), id)
}
func (s *messageService) GetConversationReadTarget(conversationID, messageID int64) (*models.Message, error) {
if messageID > 0 {
message := s.Get(messageID)
if message == nil || message.ConversationID != conversationID {
return nil, errorsx.InvalidParam("消息不存在")
}
return message, nil
}
return s.FindOne(sqls.NewCnd().Eq("conversation_id", conversationID).Desc("seq_no").Desc("id")), nil
}
func (s *messageService) SendMessage(conversationID int64, senderType enums.IMSenderType, reqSenderID int64, clientMsgID string, messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal, external *openidentity.ExternalUser) (*models.Message, error) {
switch senderType {
case enums.IMSenderTypeAgent:
return s.sendMessage(conversationID, enums.IMSenderTypeAgent, reqSenderID, clientMsgID, messageType, content, payload, operator, nil)
case enums.IMSenderTypeAI:
return s.sendMessage(conversationID, enums.IMSenderTypeAI, reqSenderID, clientMsgID, messageType, content, payload, operator, nil)
case enums.IMSenderTypeCustomer:
return s.sendMessage(conversationID, enums.IMSenderTypeCustomer, 0, clientMsgID, messageType, content, payload, nil, external)
default:
return nil, errorsx.InvalidParam("不支持的发送人类型")
}
}
func (s *messageService) SendAgentMessage(conversationID int64, reqSenderID int64, clientMsgID string, messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal) (*models.Message, error) {
return s.sendMessage(conversationID, enums.IMSenderTypeAgent, reqSenderID, clientMsgID, messageType, content, payload, operator, nil)
}
func (s *messageService) RecallAgentMessage(messageID int64, operator *dto.AuthPrincipal) (*models.Message, error) {
if operator == nil {
return nil, errorsx.Unauthorized("未登录或登录已过期")
}
if messageID <= 0 {
return nil, errorsx.InvalidParam("消息不存在")
}
message := s.Get(messageID)
if message == nil {
return nil, errorsx.InvalidParam("消息不存在")
}
if message.SenderType != enums.IMSenderTypeAgent {
return nil, errorsx.InvalidParam("仅支持撤回客服消息")
}
if message.SenderID != operator.UserID {
return nil, errorsx.Forbidden("仅允许撤回自己发送的消息")
}
if message.RecalledAt != nil || message.SendStatus == enums.IMMessageStatusRecalled {
return nil, errorsx.InvalidParam("消息已撤回")
}
conversation, err := s.ValidateConversationSender(message.ConversationID, enums.IMSenderTypeAgent, operator, nil)
if err != nil {
return nil, err
}
now := time.Now()
err = sqls.WithTransaction(func(ctx *sqls.TxContext) error {
updates := map[string]any{
"send_status": int(enums.IMMessageStatusRecalled),
"recalled_at": now,
"updated_at": now,
"update_user_id": operator.UserID,
"update_user_name": operator.Username,
}
if err := repositories.MessageRepository.Updates(ctx.Tx, message.ID, updates); err != nil {
return err
}
message.SendStatus = enums.IMMessageStatusRecalled
message.RecalledAt = &now
message.UpdatedAt = now
message.UpdateUserID = operator.UserID
message.UpdateUserName = operator.Username
agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, conversation.ID)
agentUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(agentReadState), enums.IMSenderTypeCustomer)
if err != nil {
return err
}
customerUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(customerReadState), enums.IMSenderTypeAgent, enums.IMSenderTypeAI)
if err != nil {
return err
}
conversationUpdates := map[string]any{
"agent_unread_count": agentUnreadCount,
"customer_unread_count": customerUnreadCount,
"updated_at": now,
"update_user_id": operator.UserID,
"update_user_name": operator.Username,
}
if conversation.LastMessageID == message.ID {
lastMessage := repositories.MessageRepository.FindLastUnrecalledByConversationID(ctx.Tx, conversation.ID)
if lastMessage != nil {
conversationUpdates["last_message_id"] = lastMessage.ID
conversationUpdates["last_message_at"] = lastMessage.SentAt
conversationUpdates["last_message_summary"] = limitText(buildMessageSummary(lastMessage.MessageType, lastMessage.Content), 255)
} else {
conversationUpdates["last_message_id"] = 0
conversationUpdates["last_message_at"] = nil
conversationUpdates["last_message_summary"] = ""
}
}
if err := repositories.ConversationRepository.Updates(ctx.Tx, conversation.ID, conversationUpdates); err != nil {
return err
}
if err := ConversationEventLogService.CreateEvent(ctx, conversation.ID, enums.IMEventTypeMessageRecall, enums.IMSenderTypeAgent, operator.UserID, "客服撤回消息", ""); err != nil {
return err
}
return nil
})
if err != nil {
return nil, err
}
if updatedConversation := ConversationService.Get(conversation.ID); updatedConversation != nil {
WsService.PublishMessageRecalled(updatedConversation, message)
WsService.PublishConversationChanged(updatedConversation, enums.IMRealtimeEventConversationUpdated)
}
return message, nil
}
func (s *messageService) SendAIMessage(conversationID int64, aiAgentID int64, clientMsgID string, messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal) (*models.Message, error) {
return s.sendMessage(conversationID, enums.IMSenderTypeAI, aiAgentID, clientMsgID, messageType, content, payload, operator, nil)
}
func (s *messageService) SendAIServiceNotice(conversationID int64, aiAgentID int64, content string) (*models.Message, error) {
conversation := ConversationService.Get(conversationID)
if conversation == nil {
return nil, errorsx.InvalidParam("会话不存在")
}
if conversation.Status == enums.IMConversationStatusClosed {
return nil, errorsx.InvalidParam("会话已关闭")
}
return s.sendValidatedMessage(conversation, enums.IMSenderTypeAI, aiAgentID, strs.UUID(), enums.IMMessageTypeText, content, "", &dto.AuthPrincipal{
UserID: 0,
Username: "system",
Nickname: "system",
}, nil)
}
func (s *messageService) CreateAIWelcomeMessageTx(ctx *sqls.TxContext, conversation *models.Conversation, aiAgent *models.AIAgent, now time.Time) (*models.Message, error) {
if ctx == nil || conversation == nil || aiAgent == nil || strings.TrimSpace(aiAgent.WelcomeMessage) == "" {
return nil, nil
}
content, payload, summary, err := s.normalizeMessageContent(conversation.ID, enums.IMMessageTypeText, aiAgent.WelcomeMessage, "")
if err != nil {
return nil, err
}
if strs.IsBlank(content) && strs.IsBlank(payload) {
return nil, nil
}
operator := &dto.AuthPrincipal{
UserID: 0,
Username: "system",
Nickname: "system",
}
message := &models.Message{
ConversationID: conversation.ID,
ClientMsgID: strs.UUID(),
SenderType: enums.IMSenderTypeAI,
SenderID: aiAgent.ID,
MessageType: enums.IMMessageTypeText,
Content: content,
Payload: payload,
SeqNo: repositories.MessageRepository.NextSeqNo(ctx.Tx, conversation.ID),
SendStatus: enums.IMMessageStatusSent,
SentAt: &now,
AuditFields: models.AuditFields{
CreatedAt: now,
CreateUserID: operator.UserID,
CreateUserName: operator.Username,
UpdatedAt: now,
UpdateUserID: operator.UserID,
UpdateUserName: operator.Username,
},
}
if err := repositories.MessageRepository.Create(ctx.Tx, message); err != nil {
return nil, err
}
if _, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, operator, message, now); err != nil {
return nil, err
}
agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, conversation.ID)
agentUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(agentReadState), enums.IMSenderTypeCustomer)
if err != nil {
return nil, err
}
customerUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(customerReadState), enums.IMSenderTypeAgent, enums.IMSenderTypeAI)
if err != nil {
return nil, err
}
conversationUpdates := map[string]any{
"last_message_id": message.ID,
"last_message_at": now,
"last_active_at": now,
"last_message_summary": limitText(summary, 255),
"update_user_id": operator.UserID,
"update_user_name": operator.Username,
"updated_at": now,
"agent_unread_count": agentUnreadCount,
"customer_unread_count": customerUnreadCount,
}
if err := repositories.ConversationRepository.Updates(ctx.Tx, conversation.ID, conversationUpdates); err != nil {
return nil, err
}
if err := ConversationEventLogService.CreateEvent(ctx,
conversation.ID,
enums.IMEventTypeMessageSend,
enums.IMSenderTypeAI,
0,
enums.GetIMSenderTypeLabel(enums.IMSenderTypeAI)+"发送消息",
"",
); err != nil {
return nil, err
}
conversation.LastMessageID = message.ID
conversation.LastMessageAt = now
conversation.LastActiveAt = now
conversation.LastMessageSummary = limitText(summary, 255)
conversation.AgentUnreadCount = int(agentUnreadCount)
conversation.CustomerUnreadCount = int(customerUnreadCount)
conversation.UpdatedAt = now
conversation.UpdateUserID = operator.UserID
conversation.UpdateUserName = operator.Username
return message, nil
}
func (s *messageService) SendCustomerMessage(conversationID int64, clientMsgID string, messageType enums.IMMessageType, content, payload string, external openidentity.ExternalUser) (*models.Message, error) {
ext := external
return s.sendMessage(conversationID, enums.IMSenderTypeCustomer, 0, clientMsgID, messageType, content, payload, nil, &ext)
}
func (s *messageService) sendMessage(conversationID int64, senderType enums.IMSenderType, reqSenderID int64, clientMsgID string,
messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal, external *openidentity.ExternalUser) (*models.Message, error) {
if senderType == enums.IMSenderTypeCustomer {
if external == nil || strings.TrimSpace(external.ExternalID) == "" {
return nil, errorsx.Unauthorized("外部用户标识不能为空")
}
} else if operator == nil {
return nil, errorsx.Unauthorized("未登录或登录已过期")
}
if strs.IsBlank(string(messageType)) {
messageType = enums.IMMessageTypeText
}
conversation, err := s.ValidateConversationSender(conversationID, senderType, operator, external)
if err != nil {
return nil, err
}
return s.sendValidatedMessage(conversation, senderType, reqSenderID, clientMsgID, messageType, content, payload, operator, external)
}
func (s *messageService) sendValidatedMessage(conversation *models.Conversation, senderType enums.IMSenderType, reqSenderID int64, clientMsgID string,
messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal, external *openidentity.ExternalUser) (*models.Message, error) {
var err error
var summary string
content, payload, summary, err = s.normalizeMessageContent(conversation.ID, messageType, content, payload)
if err != nil {
return nil, err
}
if strs.IsBlank(content) && strs.IsBlank(payload) {
return nil, errorsx.InvalidParam("消息内容不能为空")
}
// 防抖,消息存在就不再发送了
if strs.IsNotBlank(clientMsgID) {
if existing := repositories.MessageRepository.GetByClientMsgID(sqls.DB(), conversation.ID, clientMsgID); existing != nil {
return existing, nil
}
}
var (
now = time.Now()
auditUserID = int64(0)
auditUserName = ""
nextSeq = repositories.MessageRepository.NextSeqNo(sqls.DB(), conversation.ID)
)
if operator != nil {
auditUserID = operator.UserID
auditUserName = operator.Username
}
if senderType == enums.IMSenderTypeCustomer && external != nil {
auditUserID = 0
auditUserName = displayExternalName(external)
}
message := &models.Message{
ConversationID: conversation.ID,
ClientMsgID: clientMsgID,
SenderType: senderType,
SenderID: reqSenderID,
MessageType: messageType,
Content: content,
Payload: payload,
SeqNo: nextSeq,
SendStatus: enums.IMMessageStatusSent,
SentAt: &now,
AuditFields: models.AuditFields{
CreatedAt: now,
CreateUserID: auditUserID,
CreateUserName: auditUserName,
UpdatedAt: now,
UpdateUserID: auditUserID,
UpdateUserName: auditUserName,
},
}
switch senderType {
case enums.IMSenderTypeAgent:
if message.SenderID == 0 {
message.SenderID = operator.UserID
}
case enums.IMSenderTypeAI:
if message.SenderID == 0 {
message.SenderID = reqSenderID
}
default:
message.SenderID = 0
}
err = sqls.WithTransaction(func(ctx *sqls.TxContext) error {
if err := repositories.MessageRepository.Create(ctx.Tx, message); err != nil {
return err
}
// 处理已读、维度
readStateType := senderType
if senderType == enums.IMSenderTypeAI {
readStateType = enums.IMSenderTypeAgent
}
if readStateType == enums.IMSenderTypeAgent {
if _, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, operator, message, now); err != nil {
return err
}
} else {
if _, err := ConversationReadStateService.MarkCustomerRead(ctx, conversation, external, message, now); err != nil {
return err
}
}
agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, conversation.ID)
agentUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(agentReadState), enums.IMSenderTypeCustomer)
if err != nil {
return err
}
customerUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(customerReadState), enums.IMSenderTypeAgent, enums.IMSenderTypeAI)
if err != nil {
return err
}
updateUserID := int64(0)
updateUserName := ""
if operator != nil {
updateUserID = operator.UserID
updateUserName = operator.Username
}
if senderType == enums.IMSenderTypeCustomer && external != nil {
updateUserID = 0
updateUserName = displayExternalName(external)
}
if err := repositories.ConversationRepository.Updates(ctx.Tx, conversation.ID, map[string]any{
"last_message_id": message.ID,
"last_message_at": now,
"last_active_at": now,
"last_message_summary": limitText(summary, 255),
"update_user_id": updateUserID,
"update_user_name": updateUserName,
"updated_at": now,
"agent_unread_count": agentUnreadCount,
"customer_unread_count": customerUnreadCount,
}); err != nil {
return err
}
if err := ConversationEventLogService.CreateEvent(ctx,
conversation.ID,
enums.IMEventTypeMessageSend,
senderType,
func() int64 {
if operator != nil {
return operator.UserID
}
return 0
}(),
enums.GetIMSenderTypeLabel(senderType)+"发送消息",
"",
); err != nil {
return err
}
return nil
})
if err != nil {
return nil, err
}
WsService.PublishMessageCreated(conversation, message)
WsService.PublishConversationChanged(conversation, enums.IMRealtimeEventConversationUpdated)
if enqueueErr := ChannelMessageOutboxService.EnqueueWxWorkKFMessage(conversation, message); enqueueErr != nil {
slog.Error("enqueue wxwork kf outbox failed",
"conversation_id", conversation.ID,
"message_id", message.ID,
"error", enqueueErr,
)
}
if senderType == enums.IMSenderTypeCustomer {
if TriggerAIReplyAsyncHook != nil {
TriggerAIReplyAsyncHook(*conversation, *message)
}
}
return message, err
}
func limitText(value string, maxLen int) string {
if maxLen <= 0 {
return ""
}
value = strings.TrimSpace(value)
runes := []rune(value)
if len(runes) <= maxLen {
return value
}
return string(runes[:maxLen])
}
func buildMessageSummary(messageType enums.IMMessageType, content string) string {
content = strings.TrimSpace(content)
if content != "" {
return content
}
switch messageType {
case enums.IMMessageTypeImage:
return "[图片]"
case enums.IMMessageTypeAttachment:
return "[附件]"
case enums.IMMessageTypeHTML:
return utils.BuildHTMLSummary(content)
case "":
return ""
default:
return "[" + string(messageType) + "]"
}
}
func (s *messageService) normalizeMessageContent(conversationID int64, messageType enums.IMMessageType, content, payload string) (string, string, string, error) {
switch messageType {
case enums.IMMessageTypeHTML:
sanitized := utils.SanitizeMessageHTML(content)
normalized, err := utils.NormalizeMessageHTMLAssets(sanitized)
if err != nil {
return "", "", "", errorsx.InvalidParam("HTML消息中的图片必须使用已上传文件")
}
summary := utils.BuildHTMLSummary(normalized)
if summary == "" {
return "", "", "", errorsx.InvalidParam("消息内容不能为空")
}
return normalized, "", summary, nil
case enums.IMMessageTypeImage, enums.IMMessageTypeAttachment:
assetPayload, err := parseIMMessageAssetPayload(payload)
if err != nil {
return "", "", "", err
}
asset := AssetService.GetByAssetID(assetPayload.AssetID)
if err := validateConversationAsset(asset, conversationID, messageType); err != nil {
return "", "", "", err
}
canonicalPayload, err := buildIMMessageAssetPayload(asset)
if err != nil {
return "", "", "", err
}
summary := "[附件]"
if messageType == enums.IMMessageTypeImage {
summary = "[图片]"
}
content = strings.TrimSpace(asset.Filename)
return content, canonicalPayload, summary + s.suffixFilenameForSummary(asset.Filename), nil
default:
content = strings.TrimSpace(content)
if content == "" && strings.TrimSpace(payload) == "" {
return "", "", "", errorsx.InvalidParam("消息内容不能为空")
}
return content, strings.TrimSpace(payload), buildMessageSummary(messageType, content), nil
}
}
func (s *messageService) ValidateConversationSender(conversationID int64, senderType enums.IMSenderType, operator *dto.AuthPrincipal, external *openidentity.ExternalUser) (*models.Conversation, error) {
conversation := ConversationService.Get(conversationID)
if conversation == nil {
return nil, errorsx.InvalidParam("会话不存在")
}
if conversation.Status == enums.IMConversationStatusClosed {
return nil, errorsx.InvalidParam("会话已关闭")
}
switch senderType {
case enums.IMSenderTypeAgent:
if operator == nil {
return nil, errorsx.Unauthorized("未登录或登录已过期")
}
if conversation.Status != enums.IMConversationStatusActive || conversation.CurrentAssigneeID == 0 {
return nil, errorsx.InvalidParam("会话未分配客服,暂不允许发送消息")
}
if conversation.CurrentAssigneeID != operator.UserID {
return nil, errorsx.Forbidden("当前会话已分配给其他客服")
}
case enums.IMSenderTypeAI:
if operator == nil {
return nil, errorsx.Unauthorized("未登录或登录已过期")
}
if conversation.Status != enums.IMConversationStatusAIServing && !s.allowAIMessageOnPendingHandoff(conversation) {
return nil, errorsx.Forbidden("当前会话不处于 AI 接待状态")
}
if conversation.CurrentAssigneeID != 0 {
return nil, errorsx.Forbidden("当前会话已由人工客服接管")
}
case enums.IMSenderTypeCustomer:
if external == nil || !ConversationService.IsCustomerConversationOwner(conversation, *external) {
return nil, errorsx.Forbidden("无权访问该会话")
}
default:
return nil, errorsx.InvalidParam("不支持的发送人类型")
}
return conversation, nil
}
func (s *messageService) allowAIMessageOnPendingHandoff(conversation *models.Conversation) bool {
if conversation == nil {
return false
}
return conversation.Status == enums.IMConversationStatusPending &&
conversation.HandoffAt != nil &&
conversation.CurrentAssigneeID == 0
}
func (s *messageService) suffixFilenameForSummary(filename string) string {
filename = strings.TrimSpace(filename)
if filename == "" {
return ""
}
return " " + filename
}
func (s *messageService) readSeqNo(state *models.ConversationReadState) int64 {
if state == nil {
return 0
}
return state.LastReadSeqNo
}