From d30e21199680bf107c19d397ee2174b19fce8490 Mon Sep 17 00:00:00 2001 From: mlogclub Date: Sat, 9 May 2026 22:05:34 +0800 Subject: [PATCH] refactor: enhance conversation handling and websocket integration with new team ID support --- internal/pkg/enums/im.go | 8 +- ...nversation_human_dispatch_realtime_test.go | 242 ++++++++++++++++++ .../conversation_human_dispatch_service.go | 45 +++- .../conversation_read_state_service.go | 12 +- internal/services/conversation_service.go | 17 +- internal/services/message_service.go | 99 ++++--- internal/services/ws_realtime_types.go | 1 + internal/services/ws_service.go | 1 + web/hooks/use-agent-conversation-realtime.ts | 6 + web/lib/agent-conversation-realtime.test.mjs | 27 ++ web/lib/agent-conversation-realtime.ts | 18 ++ web/lib/api/im.ts | 1 + 12 files changed, 409 insertions(+), 68 deletions(-) create mode 100644 internal/services/conversation_human_dispatch_realtime_test.go create mode 100644 web/lib/agent-conversation-realtime.test.mjs create mode 100644 web/lib/agent-conversation-realtime.ts diff --git a/internal/pkg/enums/im.go b/internal/pkg/enums/im.go index 2edbdf6..4b7b8ad 100644 --- a/internal/pkg/enums/im.go +++ b/internal/pkg/enums/im.go @@ -54,10 +54,10 @@ func GetIMConversationServiceModeLabel(mode IMConversationServiceMode) string { type IMSenderType string const ( - IMSenderTypeAgent IMSenderType = "agent" - IMSenderTypeCustomer IMSenderType = "customer" - IMSenderTypeAI IMSenderType = "ai" - IMSenderTypeSystem IMSenderType = "system" + IMSenderTypeAgent IMSenderType = "agent" // 客服 + IMSenderTypeCustomer IMSenderType = "customer" // 客户 + IMSenderTypeAI IMSenderType = "ai" // AI + IMSenderTypeSystem IMSenderType = "system" // 系统 ) var imSenderTypeLabelMap = map[IMSenderType]string{ diff --git a/internal/services/conversation_human_dispatch_realtime_test.go b/internal/services/conversation_human_dispatch_realtime_test.go new file mode 100644 index 0000000..b95c131 --- /dev/null +++ b/internal/services/conversation_human_dispatch_realtime_test.go @@ -0,0 +1,242 @@ +package services + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "cs-agent/internal/models" + "cs-agent/internal/pkg/enums" + + "github.com/glebarez/sqlite" + "github.com/mlogclub/simple/sqls" + "gorm.io/gorm" + "gorm.io/gorm/schema" +) + +func TestAIHandoffPublishesFinalAssignedConversationEvent(t *testing.T) { + db := setupHumanDispatchRealtimeTestDB(t) + WsService = newWsService() + session := captureHumanDispatchRealtimeSession(t, "admin:101", "admin:all") + aiAgent := createHumanDispatchRealtimeAIAgent(t, db, "1") + createHumanDispatchRealtimeTeam(t, db, 1) + createHumanDispatchRealtimeActiveSchedule(t, db, 1) + createHumanDispatchRealtimeAgentProfile(t, db, 101, 1) + conversation := createHumanDispatchRealtimeConversation(t, db, aiAgent.ID) + + result, err := ConversationHumanDispatchService.HandoffByAI(conversation.ID, aiAgent, "用户要求转人工") + if err != nil { + t.Fatalf("HandoffByAI() error = %v", err) + } + if result == nil || result.Decision != HandoffDecisionAssigned { + t.Fatalf("expected assigned decision, got %+v", result) + } + + event := findHumanDispatchRealtimeEvent(t, session, enums.IMRealtimeEventConversationAssigned) + if event.Data["conversationId"] != float64(conversation.ID) { + t.Fatalf("unexpected conversation id in event: %+v", event.Data) + } + if event.Data["status"] != float64(enums.IMConversationStatusActive) { + t.Fatalf("expected active status in assigned event, got %+v", event.Data["status"]) + } + if event.Data["currentAssigneeId"] != float64(101) { + t.Fatalf("expected assignee 101 in assigned event, got %+v", event.Data["currentAssigneeId"]) + } +} + +func TestAIHandoffPublishesFinalTeamPoolConversationEvent(t *testing.T) { + db := setupHumanDispatchRealtimeTestDB(t) + WsService = newWsService() + session := captureHumanDispatchRealtimeSession(t, "admin:all") + aiAgent := createHumanDispatchRealtimeAIAgent(t, db, "1") + createHumanDispatchRealtimeTeam(t, db, 1) + createHumanDispatchRealtimeActiveSchedule(t, db, 1) + conversation := createHumanDispatchRealtimeConversation(t, db, aiAgent.ID) + + result, err := ConversationHumanDispatchService.HandoffByAI(conversation.ID, aiAgent, "用户要求转人工") + if err != nil { + t.Fatalf("HandoffByAI() error = %v", err) + } + if result == nil || result.Decision != HandoffDecisionTeamPool { + t.Fatalf("expected team_pool decision, got %+v", result) + } + + event := findHumanDispatchRealtimeEvent(t, session, enums.IMRealtimeEventConversationUpdated, func(event humanDispatchRealtimeEvent) bool { + return event.Data["currentTeamId"] == float64(1) + }) + if event.Data["conversationId"] != float64(conversation.ID) { + t.Fatalf("unexpected conversation id in event: %+v", event.Data) + } + if event.Data["status"] != float64(enums.IMConversationStatusPending) { + t.Fatalf("expected pending status in updated event, got %+v", event.Data["status"]) + } + if value, ok := event.Data["currentAssigneeId"]; ok && value != float64(0) { + t.Fatalf("expected no assignee in updated event, got %+v", event.Data["currentAssigneeId"]) + } +} + +type humanDispatchRealtimeEvent struct { + Type string `json:"type"` + Data map[string]any `json:"data"` +} + +func findHumanDispatchRealtimeEvent(t *testing.T, session *ClientSession, eventType string, matchers ...func(humanDispatchRealtimeEvent) bool) humanDispatchRealtimeEvent { + t.Helper() + timeout := time.After(500 * time.Millisecond) + for { + select { + case raw := <-session.Send: + var event humanDispatchRealtimeEvent + if err := json.Unmarshal(raw, &event); err != nil { + t.Fatalf("decode realtime event: %v", err) + } + if event.Type != eventType { + continue + } + matched := true + for _, matcher := range matchers { + if !matcher(event) { + matched = false + break + } + } + if matched { + return event + } + case <-timeout: + t.Fatalf("expected realtime event %q", eventType) + } + } +} + +func captureHumanDispatchRealtimeSession(t *testing.T, topics ...string) *ClientSession { + t.Helper() + session := &ClientSession{ + ID: "test-session", + Role: realtimeRoleAdmin, + Topics: map[string]struct{}{}, + Send: make(chan []byte, 32), + } + WsService.manager.Register(session, topics) + return session +} + +func setupHumanDispatchRealtimeTestDB(t *testing.T) *gorm.DB { + t.Helper() + dbName := strings.NewReplacer("/", "_", " ", "_").Replace(t.Name()) + db, err := gorm.Open(sqlite.Open("file:"+dbName+"?mode=memory&cache=shared"), &gorm.Config{ + NamingStrategy: schema.NamingStrategy{ + TablePrefix: "t_", + SingularTable: true, + }, + }) + if err != nil { + t.Fatalf("open sqlite error = %v", err) + } + t.Cleanup(func() { + sqlDB, err := db.DB() + if err == nil { + _ = sqlDB.Close() + } + }) + if err := db.AutoMigrate( + &models.User{}, + &models.Notification{}, + &models.Customer{}, + &models.CustomerIdentity{}, + &models.Channel{}, + &models.AIAgent{}, + &models.AgentTeam{}, + &models.AgentTeamSchedule{}, + &models.AgentProfile{}, + &models.Conversation{}, + &models.ConversationParticipant{}, + &models.ConversationAssignment{}, + &models.ConversationEventLog{}, + &models.ConversationReadState{}, + &models.Message{}, + &models.ChannelMessageOutbox{}, + ); err != nil { + t.Fatalf("auto migrate error = %v", err) + } + sqls.SetDB(db) + return db +} + +func createHumanDispatchRealtimeAIAgent(t *testing.T, db *gorm.DB, teamIDs string) models.AIAgent { + t.Helper() + item := models.AIAgent{ + Name: "测试AI", + ServiceMode: enums.IMConversationServiceModeAIFirst, + TeamIDs: teamIDs, + Status: enums.StatusOk, + } + if err := db.Create(&item).Error; err != nil { + t.Fatalf("create ai agent error = %v", err) + } + return item +} + +func createHumanDispatchRealtimeTeam(t *testing.T, db *gorm.DB, id int64) { + t.Helper() + if err := db.Create(&models.AgentTeam{ID: id, Name: "售后支持组", Status: enums.StatusOk}).Error; err != nil { + t.Fatalf("create team error = %v", err) + } +} + +func createHumanDispatchRealtimeActiveSchedule(t *testing.T, db *gorm.DB, teamID int64) { + t.Helper() + now := time.Now() + if err := db.Create(&models.AgentTeamSchedule{ + TeamID: teamID, + StartAt: now.Add(-time.Hour), + EndAt: now.Add(time.Hour), + Status: enums.StatusOk, + }).Error; err != nil { + t.Fatalf("create schedule error = %v", err) + } +} + +func createHumanDispatchRealtimeAgentProfile(t *testing.T, db *gorm.DB, userID, teamID int64) { + t.Helper() + if err := db.Create(&models.User{ + ID: userID, + Username: "agent", + Nickname: "客服", + Status: enums.StatusOk, + }).Error; err != nil { + t.Fatalf("create user error = %v", err) + } + if err := db.Create(&models.AgentProfile{ + UserID: userID, + TeamID: teamID, + AgentCode: "A001", + DisplayName: "客服", + ServiceStatus: enums.ServiceStatusIdle, + MaxConcurrentCount: 3, + AutoAssignEnabled: true, + Status: enums.StatusOk, + }).Error; err != nil { + t.Fatalf("create profile error = %v", err) + } +} + +func createHumanDispatchRealtimeConversation(t *testing.T, db *gorm.DB, aiAgentID int64) models.Conversation { + t.Helper() + now := time.Now() + item := models.Conversation{ + AIAgentID: aiAgentID, + ChannelID: 1, + CustomerID: 1, + CustomerName: "测试访客", + Status: enums.IMConversationStatusAIServing, + ServiceMode: enums.IMConversationServiceModeAIFirst, + LastMessageAt: now, + LastActiveAt: now, + } + if err := db.Create(&item).Error; err != nil { + t.Fatalf("create conversation error = %v", err) + } + return item +} diff --git a/internal/services/conversation_human_dispatch_service.go b/internal/services/conversation_human_dispatch_service.go index 74e7a22..93fb87f 100644 --- a/internal/services/conversation_human_dispatch_service.go +++ b/internal/services/conversation_human_dispatch_service.go @@ -121,6 +121,7 @@ func (s *conversationHumanDispatchService) DispatchPendingConversation(conversat return nil, err } if dispatched != nil { + WsService.PublishConversationChanged(dispatched, enums.IMRealtimeEventConversationAssigned) return &HandoffDecisionResult{ Decision: HandoffDecisionAssigned, TeamID: dispatched.CurrentTeamID, @@ -129,9 +130,13 @@ func (s *conversationHumanDispatchService) DispatchPendingConversation(conversat } } teamID := activeTeamIDs[0] - if err := s.moveToTeamPool(conversationID, teamID, "手动触发自动分配"); err != nil { + teamPoolConversation, err := s.moveToTeamPool(conversationID, teamID, "手动触发自动分配") + if err != nil { return nil, err } + if teamPoolConversation != nil { + WsService.PublishConversationChanged(teamPoolConversation, enums.IMRealtimeEventConversationUpdated) + } return &HandoffDecisionResult{Decision: HandoffDecisionTeamPool, TeamID: teamID}, nil } @@ -150,6 +155,7 @@ func (s *conversationHumanDispatchService) dispatchAfterHandoff(conversationID, return nil, err } if dispatched != nil { + WsService.PublishConversationChanged(dispatched, enums.IMRealtimeEventConversationAssigned) if publishAssignEvent { eventbus.PublishAsync(context.Background(), events.ConversationAssignedEvent{ ConversationID: dispatched.ID, @@ -169,9 +175,13 @@ func (s *conversationHumanDispatchService) dispatchAfterHandoff(conversationID, } teamID := activeTeamIDs[0] - if err := s.moveToTeamPool(conversationID, teamID, reason); err != nil { + teamPoolConversation, err := s.moveToTeamPool(conversationID, teamID, reason) + if err != nil { return nil, err } + if teamPoolConversation != nil { + WsService.PublishConversationChanged(teamPoolConversation, enums.IMRealtimeEventConversationUpdated) + } return &HandoffDecisionResult{Decision: HandoffDecisionTeamPool, TeamID: teamID, Message: HandoffWaitingMessage}, nil } @@ -195,11 +205,12 @@ func (s *conversationHumanDispatchService) markHandoff(conversationID int64, aiA }) } -func (s *conversationHumanDispatchService) moveToTeamPool(conversationID, teamID int64, reason string) error { +func (s *conversationHumanDispatchService) moveToTeamPool(conversationID, teamID int64, reason string) (*models.Conversation, error) { now := time.Now() - return sqls.WithTransaction(func(ctx *sqls.TxContext) error { - conversation := repositories.ConversationRepository.Get(ctx.Tx, conversationID) - if conversation == nil { + var conversation *models.Conversation + err := sqls.WithTransaction(func(ctx *sqls.TxContext) error { + current := repositories.ConversationRepository.Get(ctx.Tx, conversationID) + if current == nil { return errorsx.InvalidParam("会话不存在") } if err := ConversationAssignmentService.FinishActiveAssignments(ctx, conversationID, now); err != nil { @@ -215,16 +226,30 @@ func (s *conversationHumanDispatchService) moveToTeamPool(conversationID, teamID }); err != nil { return err } - return ConversationEventLogService.CreateEvent(ctx, conversationID, enums.IMEventTypeTransfer, enums.IMSenderTypeSystem, 0, "会话进入客服组待接入", ConversationService.buildEventPayload(map[string]any{ - "fromStatus": conversation.Status, + if err := ConversationEventLogService.CreateEvent(ctx, conversationID, enums.IMEventTypeTransfer, enums.IMSenderTypeSystem, 0, "会话进入客服组待接入", ConversationService.buildEventPayload(map[string]any{ + "fromStatus": current.Status, "toStatus": enums.IMConversationStatusPending, - "fromAssigneeId": conversation.CurrentAssigneeID, + "fromAssigneeId": current.CurrentAssigneeID, "toAssigneeId": int64(0), "toTeamId": teamID, "reason": strings.TrimSpace(reason), "decision": string(HandoffDecisionTeamPool), - })) + })); err != nil { + return err + } + current.Status = enums.IMConversationStatusPending + current.CurrentTeamID = teamID + current.CurrentAssigneeID = 0 + current.UpdateUserID = 0 + current.UpdateUserName = "system" + current.UpdatedAt = now + conversation = current + return nil }) + if err != nil { + return nil, err + } + return conversation, nil } func (s *conversationHumanDispatchService) moveToGlobalPool(conversationID int64, operatorName string) error { diff --git a/internal/services/conversation_read_state_service.go b/internal/services/conversation_read_state_service.go index 8634c35..a205bf2 100644 --- a/internal/services/conversation_read_state_service.go +++ b/internal/services/conversation_read_state_service.go @@ -172,24 +172,24 @@ func (s *conversationReadStateService) pickConversationReadStates(list []models. } // MarkAgentRead 在事务内更新/创建客服已读游标。 -func (s *conversationReadStateService) MarkAgentRead(ctx *sqls.TxContext, conversation *models.Conversation, operator *dto.AuthPrincipal, message *models.Message, now time.Time) (*models.ConversationReadState, error) { +func (s *conversationReadStateService) MarkAgentRead(ctx *sqls.TxContext, conversation *models.Conversation, operator *dto.AuthPrincipal, message *models.Message) (*models.ConversationReadState, error) { c, err := agentReaderCursor(operator) if err != nil { return nil, err } - return s.markReadTxWithCursor(ctx, conversation, c, message, now) + return s.markReadTxWithCursor(ctx, conversation, c, message) } // MarkCustomerRead 在事务内更新/创建 IM 客户已读游标。 -func (s *conversationReadStateService) MarkCustomerRead(ctx *sqls.TxContext, conversation *models.Conversation, external *openidentity.ExternalUser, message *models.Message, now time.Time) (*models.ConversationReadState, error) { +func (s *conversationReadStateService) MarkCustomerRead(ctx *sqls.TxContext, conversation *models.Conversation, external *openidentity.ExternalUser, message *models.Message) (*models.ConversationReadState, error) { c, err := customerReaderCursor(external) if err != nil { return nil, err } - return s.markReadTxWithCursor(ctx, conversation, c, message, now) + return s.markReadTxWithCursor(ctx, conversation, c, message) } -func (s *conversationReadStateService) markReadTxWithCursor(ctx *sqls.TxContext, conversation *models.Conversation, c readerCursor, message *models.Message, now time.Time) (*models.ConversationReadState, error) { +func (s *conversationReadStateService) markReadTxWithCursor(ctx *sqls.TxContext, conversation *models.Conversation, c readerCursor, message *models.Message) (*models.ConversationReadState, error) { if ctx == nil || conversation == nil || message == nil { return nil, nil } @@ -197,6 +197,8 @@ func (s *conversationReadStateService) markReadTxWithCursor(ctx *sqls.TxContext, return nil, errorsx.InvalidParam("不支持的已读操作类型") } + now := time.Now() + item := &models.ConversationReadState{} err := ctx.Tx.Where("conversation_id = ? AND reader_type = ? AND reader_id = ? AND external_reader_id = ?", conversation.ID, c.readerType, c.readerID, c.externalReaderID, diff --git a/internal/services/conversation_service.go b/internal/services/conversation_service.go index 3627a98..833034f 100644 --- a/internal/services/conversation_service.go +++ b/internal/services/conversation_service.go @@ -153,7 +153,7 @@ func (s *conversationService) Create(externalUser openidentity.ExternalUser, cha if err := ConversationEventLogService.CreateEvent(ctx, conversation.ID, enums.IMEventTypeCreate, enums.IMSenderTypeCustomer, 0, "用户创建会话", ""); err != nil { return err } - welcomeMessage, err = MessageService.CreateAIWelcomeMessageTx(ctx, conversation, aiAgent, now) + welcomeMessage, err = MessageService.createAIWelcomeMessage(ctx, conversation, aiAgent, now) return err }); err != nil { return nil, err @@ -517,7 +517,7 @@ func displayExternalName(ext *openidentity.ExternalUser) string { type conversationReadActor interface { isAgentSide() bool getReadState(conversationID int64) *models.ConversationReadState - markReadTx(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message, now time.Time) error + markRead(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message) error conversationUpdateAudit() (userID int64, userName string) } @@ -531,8 +531,8 @@ func (a agentConversationReadActor) getReadState(conversationID int64) *models.C return ConversationReadStateService.GetByAgentReader(conversationID, a.operator) } -func (a agentConversationReadActor) markReadTx(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message, now time.Time) error { - _, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, a.operator, targetMessage, now) +func (a agentConversationReadActor) markRead(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message) error { + _, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, a.operator, targetMessage) return err } @@ -553,8 +553,8 @@ func (a customerConversationReadActor) getReadState(conversationID int64) *model return ConversationReadStateService.GetByCustomerReader(conversationID, a.external) } -func (a customerConversationReadActor) markReadTx(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message, now time.Time) error { - _, err := ConversationReadStateService.MarkCustomerRead(ctx, conversation, a.external, targetMessage, now) +func (a customerConversationReadActor) markRead(ctx *sqls.TxContext, conversation *models.Conversation, targetMessage *models.Message) error { + _, err := ConversationReadStateService.MarkCustomerRead(ctx, conversation, a.external, targetMessage) return err } @@ -607,8 +607,7 @@ func (s *conversationService) markConversationReadWithActor(conversation *models if currentConversation == nil { return errorsx.InvalidParam("会话不存在") } - now := time.Now() - if err := actor.markReadTx(ctx, currentConversation, targetMessage, now); err != nil { + if err := actor.markRead(ctx, currentConversation, targetMessage); err != nil { return err } agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, currentConversation.ID) @@ -632,7 +631,7 @@ func (s *conversationService) markConversationReadWithActor(conversation *models "customer_unread_count": customerUnreadCount, "update_user_id": updateUserID, "update_user_name": updateUserName, - "updated_at": now, + "updated_at": time.Now(), }) }) if err != nil { diff --git a/internal/services/message_service.go b/internal/services/message_service.go index 6106175..1ccbab2 100644 --- a/internal/services/message_service.go +++ b/internal/services/message_service.go @@ -258,7 +258,7 @@ func (s *messageService) SendAIServiceNotice(conversationID int64, aiAgentID int }, nil) } -func (s *messageService) CreateAIWelcomeMessageTx(ctx *sqls.TxContext, conversation *models.Conversation, aiAgent *models.AIAgent, now time.Time) (*models.Message, error) { +func (s *messageService) createAIWelcomeMessage(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 } @@ -300,7 +300,7 @@ func (s *messageService) CreateAIWelcomeMessageTx(ctx *sqls.TxContext, conversat return nil, err } - if _, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, operator, message, now); err != nil { + if _, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, operator, message); err != nil { return nil, err } agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, conversation.ID) @@ -450,56 +450,44 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation, } // 处理已读、维度 - 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) + agentUnreadCount, customerUnreadCount, err := s.handleReadState(ctx, senderType, conversation, operator, message, external) if err != nil { return err } - updateUserID := int64(0) - updateUserName := "" + conversation.LastMessageID = message.ID + conversation.LastMessageAt = now + conversation.LastActiveAt = now + conversation.LastMessageSummary = limitText(summary, 255) + conversation.UpdateUserID = int64(0) + conversation.UpdateUserName = "" if operator != nil { - updateUserID = operator.UserID - updateUserName = operator.Username + conversation.UpdateUserID = operator.UserID + conversation.UpdateUserName = operator.Username } if senderType == enums.IMSenderTypeCustomer && external != nil { - updateUserID = 0 - updateUserName = displayExternalName(external) + conversation.UpdateUserID = 0 + conversation.UpdateUserName = displayExternalName(external) } + conversation.UpdatedAt = now + conversation.AgentUnreadCount = int(agentUnreadCount) + conversation.CustomerUnreadCount = int(customerUnreadCount) 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, + "last_message_id": conversation.LastMessageID, + "last_message_at": conversation.LastMessageAt, + "last_active_at": conversation.LastActiveAt, + "last_message_summary": conversation.LastMessageSummary, + "update_user_id": conversation.UpdateUserID, + "update_user_name": conversation.UpdateUserName, + "updated_at": conversation.UpdatedAt, + "agent_unread_count": conversation.AgentUnreadCount, + "customer_unread_count": conversation.CustomerUnreadCount, }); err != nil { return err } - if err := ConversationEventLogService.CreateEvent(ctx, - conversation.ID, - enums.IMEventTypeMessageSend, - senderType, + + // 记录事件日志 + if err := ConversationEventLogService.CreateEvent(ctx, conversation.ID, enums.IMEventTypeMessageSend, senderType, func() int64 { if operator != nil { return operator.UserID @@ -511,14 +499,18 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation, ); err != nil { return err } + return nil }) if err != nil { return nil, err } + // 处理websocket消息 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, @@ -526,6 +518,8 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation, "error", enqueueErr, ) } + + // 客户发送消息，触发AI回复 if senderType == enums.IMSenderTypeCustomer { if TriggerAIReplyAsyncHook != nil { TriggerAIReplyAsyncHook(*conversation, *message) @@ -534,6 +528,31 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation, return message, err } +// handleReadState 根据发送者类型更新会话已读状态，并返回更新后的客服和客户未读消息数。 +func (s *messageService) handleReadState(ctx *sqls.TxContext, senderType enums.IMSenderType, conversation *models.Conversation, operator *dto.AuthPrincipal, message *models.Message, external *openidentity.ExternalUser) (agentUnreadCount int64, customerUnreadCount int64, err error) { + readStateType := senderType + if senderType == enums.IMSenderTypeAI { + readStateType = enums.IMSenderTypeAgent + } + if readStateType == enums.IMSenderTypeAgent { + if _, err := ConversationReadStateService.MarkAgentRead(ctx, conversation, operator, message); err != nil { + return 0, 0, err + } + } else { + if _, err := ConversationReadStateService.MarkCustomerRead(ctx, conversation, external, message); err != nil { + return 0, 0, err + } + } + agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, conversation.ID) + if agentUnreadCount, err = ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(agentReadState), enums.IMSenderTypeCustomer); err != nil { + return 0, 0, err + } + if customerUnreadCount, err = ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(customerReadState), enums.IMSenderTypeAgent, enums.IMSenderTypeAI); err != nil { + return 0, 0, err + } + return agentUnreadCount, customerUnreadCount, nil +} + func limitText(value string, maxLen int) string { if maxLen <= 0 { return "" diff --git a/internal/services/ws_realtime_types.go b/internal/services/ws_realtime_types.go index 5c73a7e..2370e72 100644 --- a/internal/services/ws_realtime_types.go +++ b/internal/services/ws_realtime_types.go @@ -192,6 +192,7 @@ type RealtimeConversationChangedPayload struct { Status enums.IMConversationStatus `json:"status,omitempty"` ServiceMode enums.IMConversationServiceMode `json:"serviceMode,omitempty"` CurrentAssigneeID int64 `json:"currentAssigneeId,omitempty"` + CurrentTeamID int64 `json:"currentTeamId,omitempty"` LastMessageID int64 `json:"lastMessageId,omitempty"` LastMessageAt string `json:"lastMessageAt,omitempty"` LastActiveAt string `json:"lastActiveAt,omitempty"` diff --git a/internal/services/ws_service.go b/internal/services/ws_service.go index 93f3ee8..430c57e 100644 --- a/internal/services/ws_service.go +++ b/internal/services/ws_service.go @@ -426,6 +426,7 @@ func (s *wsService) PublishConversationChanged(conversation *models.Conversation Status: conversation.Status, ServiceMode: conversation.ServiceMode, CurrentAssigneeID: conversation.CurrentAssigneeID, + CurrentTeamID: conversation.CurrentTeamID, LastMessageID: conversation.LastMessageID, LastMessageAt: formatWsTime(&conversation.LastMessageAt), LastActiveAt: formatWsTime(&conversation.LastActiveAt), diff --git a/web/hooks/use-agent-conversation-realtime.ts b/web/hooks/use-agent-conversation-realtime.ts index f5e992c..2980b3c 100644 --- a/web/hooks/use-agent-conversation-realtime.ts +++ b/web/hooks/use-agent-conversation-realtime.ts @@ -5,6 +5,7 @@ import { toast } from "sonner" import { createAdminWebSocketUrl } from "@/lib/api/admin" import { type AgentMessage } from "@/lib/api/agent" +import { shouldReloadConversationListForRealtimePatch } from "@/lib/agent-conversation-realtime" import { readSession } from "@/lib/auth" import { normalizeRealtimeMessage, @@ -131,6 +132,11 @@ export function useAgentConversationRealtime() { if (eventType.startsWith("conversation.") && payload) { store.applyRealtimeConversationChanged(payload) + if (shouldReloadConversationListForRealtimePatch(payload)) { + void store.resyncRealtimeData(conversationId).catch((error) => { + toast.error(error instanceof Error ? error.message : "同步会话列表失败") + }) + } } } catch { // ignore invalid ws payload diff --git a/web/lib/agent-conversation-realtime.test.mjs b/web/lib/agent-conversation-realtime.test.mjs new file mode 100644 index 0000000..f2b0290 --- /dev/null +++ b/web/lib/agent-conversation-realtime.test.mjs @@ -0,0 +1,27 @@ +import assert from "node:assert/strict" +import test from "node:test" + +import { shouldReloadConversationListForRealtimePatch } from "./agent-conversation-realtime.ts" + +test("reloads conversation list when realtime patch changes list membership fields", () => { + assert.equal( + shouldReloadConversationListForRealtimePatch({ + conversationId: 1, + status: 3, + currentAssigneeId: 101, + }), + true + ) +}) + +test("keeps local patching for message summary and unread-only changes", () => { + assert.equal( + shouldReloadConversationListForRealtimePatch({ + conversationId: 1, + lastMessageId: 9, + lastMessageSummary: "hello", + agentUnreadCount: 1, + }), + false + ) +}) diff --git a/web/lib/agent-conversation-realtime.ts b/web/lib/agent-conversation-realtime.ts new file mode 100644 index 0000000..bb0e043 --- /dev/null +++ b/web/lib/agent-conversation-realtime.ts @@ -0,0 +1,18 @@ +import type { RealtimeConversationPatch } from "@/lib/im-realtime-state" + +const listMembershipFields = new Set([ + "status", + "currentAssigneeId", + "currentTeamId", +]) + +export function shouldReloadConversationListForRealtimePatch( + patch: RealtimeConversationPatch | null | undefined +) { + if (!patch) { + return false + } + return Object.keys(patch).some((key) => + listMembershipFields.has(key as keyof RealtimeConversationPatch) + ) +} diff --git a/web/lib/api/im.ts b/web/lib/api/im.ts index 7297b5d..6a828aa 100644 --- a/web/lib/api/im.ts +++ b/web/lib/api/im.ts @@ -40,6 +40,7 @@ export type ImConversation = { priority: number currentAssigneeId: number currentAssigneeName?: string + currentTeamId?: number lastMessageId: number lastMessageAt?: string lastActiveAt?: string