From 85cfee19676ce597aeff6e3d3ce0d6b286036639 Mon Sep 17 00:00:00 2001 From: mlogclub Date: Wed, 24 Jun 2026 22:53:17 +0800 Subject: [PATCH] refactor: remove SeqNo from message and conversation read state models, update related logic --- docs | 2 +- internal/ai/runtime/workflow/executor_test.go | 1 - internal/builders/conversation_builder.go | 12 +- .../builders/conversation_builder_test.go | 20 +++ internal/models/models.go | 4 +- .../pkg/dto/response/conversation_response.go | 2 - internal/pkg/dto/response/message_response.go | 1 - internal/repositories/message_repository.go | 9 -- .../conversation_read_state_service.go | 13 +- internal/services/conversation_service.go | 12 +- internal/services/message_service.go | 21 ++- internal/services/message_service_test.go | 133 +++++++++++++++++- internal/services/ws_realtime_types.go | 3 - internal/services/ws_service.go | 17 +-- 14 files changed, 177 insertions(+), 73 deletions(-) diff --git a/docs b/docs index db4fd85..c95d51e 160000 --- a/docs +++ b/docs @@ -1 +1 @@ -Subproject commit db4fd857c48e065128c51f57e8eded72780d91f3 +Subproject commit c95d51e0b92a8d42060d5d7832f2097a9428bc50 diff --git a/internal/ai/runtime/workflow/executor_test.go b/internal/ai/runtime/workflow/executor_test.go index d3db7e3..91a662d 100644 --- a/internal/ai/runtime/workflow/executor_test.go +++ b/internal/ai/runtime/workflow/executor_test.go @@ -698,7 +698,6 @@ func createWorkflowExecutorCustomerMessage(t *testing.T, db *gorm.DB, conversati SenderType: enums.IMSenderTypeCustomer, MessageType: enums.IMMessageTypeText, Content: content, - SeqNo: 1, SendStatus: enums.IMMessageStatusSent, SentAt: &now, } diff --git a/internal/builders/conversation_builder.go b/internal/builders/conversation_builder.go index b570cf7..0846aec 100644 --- a/internal/builders/conversation_builder.go +++ b/internal/builders/conversation_builder.go @@ -37,10 +37,8 @@ func BuildConversationWithLocale(item *models.Conversation, locale string) respo CustomerUnreadCount: item.CustomerUnreadCount, AgentUnreadCount: item.AgentUnreadCount, CustomerLastReadMessageID: readStateMessageID(customerReadState), - CustomerLastReadSeqNo: readStateSeqNo(customerReadState), CustomerLastReadAt: readStateAt(customerReadState), AgentLastReadMessageID: readStateMessageID(agentReadState), - AgentLastReadSeqNo: readStateSeqNo(agentReadState), AgentLastReadAt: readStateAt(agentReadState), ClosedAt: utils.FormatTimePtr(item.ClosedAt), ClosedBy: item.ClosedBy, @@ -157,7 +155,6 @@ func BuildMessageWithReadStatesAndLocale(item *models.Message, agentReadState, c MessageType: item.MessageType, Content: localizeRenderableMessageContent(locale, content), Payload: payload, - SeqNo: item.SeqNo, SendStatus: item.SendStatus, SentAt: utils.FormatTimePtr(item.SentAt), DeliveredAt: utils.FormatTimePtr(item.DeliveredAt), @@ -282,7 +279,7 @@ func collectMessageSenderNameMaps(list []models.Message) (aiNames map[int64]stri } func isMessageRead(item *models.Message, state *models.ConversationReadState) bool { - return item != nil && state != nil && state.LastReadSeqNo >= item.SeqNo + return item != nil && state != nil && state.LastReadMessageID >= item.ID } func readMessageAt(item *models.Message, state *models.ConversationReadState) string { @@ -299,13 +296,6 @@ func readStateMessageID(state *models.ConversationReadState) int64 { return state.LastReadMessageID } -func readStateSeqNo(state *models.ConversationReadState) int64 { - if state == nil { - return 0 - } - return state.LastReadSeqNo -} - func readStateAt(state *models.ConversationReadState) string { if state == nil { return "" diff --git a/internal/builders/conversation_builder_test.go b/internal/builders/conversation_builder_test.go index 86e1c6f..78cc802 100644 --- a/internal/builders/conversation_builder_test.go +++ b/internal/builders/conversation_builder_test.go @@ -1,6 +1,8 @@ package builders import ( + "encoding/json" + "strings" "testing" "agent-desk/internal/models" @@ -114,3 +116,21 @@ func TestBuildMessageIncludesWorkflowRunID(t *testing.T) { t.Fatalf("resp.WorkflowRunID=%d want 9988", resp.WorkflowRunID) } } + +func TestBuildMessageJSONDoesNotExposeSeqNo(t *testing.T) { + resp := BuildMessageWithReadStatesAndLocale(&models.Message{ + ID: 1, + ConversationID: 2, + SenderType: enums.IMSenderTypeCustomer, + MessageType: enums.IMMessageTypeText, + Content: "hello", + }, nil, nil, nil, nil, nil, i18nx.DefaultLocale) + + raw, err := json.Marshal(resp) + if err != nil { + t.Fatalf("marshal message response: %v", err) + } + if strings.Contains(string(raw), "seqNo") { + t.Fatalf("message response should not expose seqNo, got %s", raw) + } +} diff --git a/internal/models/models.go b/internal/models/models.go index e6b212e..3051186 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -380,7 +380,6 @@ type ConversationReadState struct { ReaderID int64 `gorm:"type:bigint;not null;default:0;uniqueIndex:uk_conversation_reader"` ExternalReaderID string `gorm:"type:varchar(128);not null;default:'';uniqueIndex:uk_conversation_reader"` LastReadMessageID int64 `gorm:"type:bigint;not null;default:0;index"` - LastReadSeqNo int64 `gorm:"type:bigint;not null;default:0;index"` LastReadAt *time.Time `gorm:"type:datetime"` AuditFields } @@ -388,7 +387,7 @@ type ConversationReadState struct { // Message 会话消息。 type Message struct { ID int64 `gorm:"primaryKey;autoIncrement"` - ConversationID int64 `gorm:"type:bigint;not null;index;uniqueIndex:uk_conversation_seq;uniqueIndex:uk_conversation_client_msg"` + ConversationID int64 `gorm:"type:bigint;not null;index;uniqueIndex:uk_conversation_client_msg"` RequestID string `gorm:"type:varchar(128);not null;default:'';index"` WorkflowRunID int64 `gorm:"type:bigint;not null;default:0;index"` ClientMsgID string `gorm:"type:varchar(128);not null;default:'';uniqueIndex:uk_conversation_client_msg"` @@ -398,7 +397,6 @@ type Message struct { MessageType enums.IMMessageType `gorm:"type:varchar(30);not null;default:'';index"` Content string `gorm:"type:text"` Payload string `gorm:"type:text"` - SeqNo int64 `gorm:"type:bigint;not null;default:0;uniqueIndex:uk_conversation_seq"` SendStatus enums.IMMessageStatus `gorm:"type:int;not null;default:2;index"` SentAt *time.Time `gorm:"type:datetime;index"` DeliveredAt *time.Time `gorm:"type:datetime"` diff --git a/internal/pkg/dto/response/conversation_response.go b/internal/pkg/dto/response/conversation_response.go index f1b67c7..697dbfd 100644 --- a/internal/pkg/dto/response/conversation_response.go +++ b/internal/pkg/dto/response/conversation_response.go @@ -37,10 +37,8 @@ type ConversationResponse struct { CustomerUnreadCount int `json:"customerUnreadCount"` AgentUnreadCount int `json:"agentUnreadCount"` CustomerLastReadMessageID int64 `json:"customerLastReadMessageId"` - CustomerLastReadSeqNo int64 `json:"customerLastReadSeqNo"` CustomerLastReadAt string `json:"customerLastReadAt,omitempty"` AgentLastReadMessageID int64 `json:"agentLastReadMessageId"` - AgentLastReadSeqNo int64 `json:"agentLastReadSeqNo"` AgentLastReadAt string `json:"agentLastReadAt,omitempty"` CustomerOnline bool `json:"customerOnline"` ClosedAt string `json:"closedAt,omitempty"` diff --git a/internal/pkg/dto/response/message_response.go b/internal/pkg/dto/response/message_response.go index 349b58a..748afc2 100644 --- a/internal/pkg/dto/response/message_response.go +++ b/internal/pkg/dto/response/message_response.go @@ -15,7 +15,6 @@ type MessageResponse struct { MessageType enums.IMMessageType `json:"messageType"` Content string `json:"content"` Payload string `json:"payload,omitempty"` - SeqNo int64 `json:"seqNo"` SendStatus enums.IMMessageStatus `json:"sendStatus"` SentAt string `json:"sentAt,omitempty"` DeliveredAt string `json:"deliveredAt,omitempty"` diff --git a/internal/repositories/message_repository.go b/internal/repositories/message_repository.go index 057bdb0..5f9d0a9 100644 --- a/internal/repositories/message_repository.go +++ b/internal/repositories/message_repository.go @@ -43,7 +43,6 @@ func (r *messageRepository) FindLastUnrecalledByConversationID(db *gorm.DB, conv ret := &models.Message{} if err := db. Where("conversation_id = ? AND recalled_at IS NULL AND send_status <> ?", conversationID, 6). - Order("seq_no DESC"). Order("id DESC"). Limit(1). Take(ret).Error; err != nil { @@ -118,11 +117,3 @@ func (r *messageRepository) Delete(db *gorm.DB, id int64) { func (r *messageRepository) GetByClientMsgID(db *gorm.DB, conversationID int64, clientMsgID string) *models.Message { return r.FindOne(db, sqls.NewCnd().Where("conversation_id = ? AND client_msg_id = ?", conversationID, clientMsgID)) } - -// NextSeqNo -func (r *messageRepository) NextSeqNo(db *gorm.DB, conversationID int64) int64 { - if last := r.FindOne(db, sqls.NewCnd().Where("conversation_id = ?", conversationID).Desc("seq_no")); last != nil { - return last.SeqNo + 1 - } - return 1 -} diff --git a/internal/services/conversation_read_state_service.go b/internal/services/conversation_read_state_service.go index 2c68229..cd7bd7f 100644 --- a/internal/services/conversation_read_state_service.go +++ b/internal/services/conversation_read_state_service.go @@ -160,11 +160,11 @@ func (s *conversationReadStateService) pickConversationReadStates(list []models. item := &list[i] switch item.ReaderType { case enums.IMSenderTypeAgent: - if agentState == nil || item.LastReadSeqNo > agentState.LastReadSeqNo { + if agentState == nil || item.LastReadMessageID > agentState.LastReadMessageID { agentState = item } case enums.IMSenderTypeCustomer: - if customerState == nil || item.LastReadSeqNo > customerState.LastReadSeqNo { + if customerState == nil || item.LastReadMessageID > customerState.LastReadMessageID { customerState = item } } @@ -214,7 +214,6 @@ func (s *conversationReadStateService) markReadTxWithCursor(ctx *sqls.TxContext, ReaderID: c.readerID, ExternalReaderID: c.externalReaderID, LastReadMessageID: message.ID, - LastReadSeqNo: message.SeqNo, LastReadAt: &now, AuditFields: models.AuditFields{ CreatedAt: now, @@ -231,19 +230,17 @@ func (s *conversationReadStateService) markReadTxWithCursor(ctx *sqls.TxContext, return item, nil } - if item.LastReadSeqNo >= message.SeqNo { + if item.LastReadMessageID >= message.ID { return item, nil } item.LastReadMessageID = message.ID - item.LastReadSeqNo = message.SeqNo item.LastReadAt = &now item.UpdatedAt = now item.UpdateUserID = c.auditUserID item.UpdateUserName = c.auditUserName if err := repositories.ConversationReadStateRepository.Updates(ctx.Tx, item.ID, map[string]any{ "last_read_message_id": item.LastReadMessageID, - "last_read_seq_no": item.LastReadSeqNo, "last_read_at": item.LastReadAt, "updated_at": item.UpdatedAt, "update_user_id": item.UpdateUserID, @@ -254,7 +251,7 @@ func (s *conversationReadStateService) markReadTxWithCursor(ctx *sqls.TxContext, return item, nil } -func (s *conversationReadStateService) CountUnreadMessages(ctx *sqls.TxContext, conversationID, lastReadSeqNo int64, senderTypes ...enums.IMSenderType) (int64, error) { +func (s *conversationReadStateService) CountUnreadMessages(ctx *sqls.TxContext, conversationID, lastReadMessageID int64, senderTypes ...enums.IMSenderType) (int64, error) { normalizedSenderTypes := make([]enums.IMSenderType, 0, len(senderTypes)) for _, senderType := range senderTypes { if strs.IsBlank(string(senderType)) { @@ -267,7 +264,7 @@ func (s *conversationReadStateService) CountUnreadMessages(ctx *sqls.TxContext, } var count int64 query := ctx.Tx.Model(&models.Message{}). - Where("conversation_id = ? AND seq_no > ? AND recalled_at IS NULL AND send_status <> ?", conversationID, lastReadSeqNo, int(enums.IMMessageStatusRecalled)) + Where("conversation_id = ? AND id > ? AND recalled_at IS NULL AND send_status <> ?", conversationID, lastReadMessageID, int(enums.IMMessageStatusRecalled)) if len(normalizedSenderTypes) == 1 { query = query.Where("sender_type = ?", normalizedSenderTypes[0]) } else { diff --git a/internal/services/conversation_service.go b/internal/services/conversation_service.go index 7c790a6..b4db806 100644 --- a/internal/services/conversation_service.go +++ b/internal/services/conversation_service.go @@ -603,7 +603,7 @@ func (s *conversationService) markConversationReadWithActor(conversation *models } currentReadState := actor.getReadState(conversation.ID) - if currentReadState != nil && currentReadState.LastReadSeqNo >= targetMessage.SeqNo { + if currentReadState != nil && currentReadState.LastReadMessageID >= targetMessage.ID { if actor.isAgentSide() && conversation.AgentUnreadCount == 0 { return false, nil } @@ -629,10 +629,10 @@ func (s *conversationService) markConversationReadWithActor(conversation *models if err != nil { return err } - if actor.isAgentSide() && currentConversation.AgentUnreadCount == agentUnreadCount && currentReadState != nil && currentReadState.LastReadSeqNo >= targetMessage.SeqNo { + if actor.isAgentSide() && currentConversation.AgentUnreadCount == agentUnreadCount && currentReadState != nil && currentReadState.LastReadMessageID >= targetMessage.ID { return nil } - if !actor.isAgentSide() && currentConversation.CustomerUnreadCount == customerUnreadCount && currentReadState != nil && currentReadState.LastReadSeqNo >= targetMessage.SeqNo { + if !actor.isAgentSide() && currentConversation.CustomerUnreadCount == customerUnreadCount && currentReadState != nil && currentReadState.LastReadMessageID >= targetMessage.ID { return nil } updateUserID, updateUserName := actor.conversationUpdateAudit() @@ -651,15 +651,15 @@ func (s *conversationService) markConversationReadWithActor(conversation *models } func (s *conversationService) countUnreadByState(ctx *sqls.TxContext, conversationID int64, state *models.ConversationReadState, senderTypes ...enums.IMSenderType) (int, error) { - lastReadSeqNo := int64(0) + lastReadMessageID := int64(0) if state != nil { - lastReadSeqNo = state.LastReadSeqNo + lastReadMessageID = state.LastReadMessageID } normalizedSenderTypes := make([]enums.IMSenderType, 0, len(senderTypes)) for _, senderType := range senderTypes { normalizedSenderTypes = append(normalizedSenderTypes, senderType) } - count, err := ConversationReadStateService.CountUnreadMessages(ctx, conversationID, lastReadSeqNo, normalizedSenderTypes...) + count, err := ConversationReadStateService.CountUnreadMessages(ctx, conversationID, lastReadMessageID, normalizedSenderTypes...) return int(count), err } diff --git a/internal/services/message_service.go b/internal/services/message_service.go index 3c2cd26..94c9e7a 100644 --- a/internal/services/message_service.go +++ b/internal/services/message_service.go @@ -126,7 +126,7 @@ func (s *messageService) GetConversationReadTarget(conversationID, messageID int } return message, nil } - return s.FindOne(sqls.NewCnd().Eq("conversation_id", conversationID).Desc("seq_no").Desc("id")), nil + return s.FindOne(sqls.NewCnd().Eq("conversation_id", conversationID).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) { @@ -197,11 +197,11 @@ func (s *messageService) RecallAgentMessage(messageID int64, operator *dto.AuthP message.UpdateUserName = operator.Username agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, conversation.ID) - agentUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(agentReadState), enums.IMSenderTypeCustomer) + agentUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readMessageID(agentReadState), enums.IMSenderTypeCustomer) if err != nil { return err } - customerUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(customerReadState), enums.IMSenderTypeAgent, enums.IMSenderTypeAI) + customerUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readMessageID(customerReadState), enums.IMSenderTypeAgent, enums.IMSenderTypeAI) if err != nil { return err } @@ -302,7 +302,6 @@ func (s *messageService) createAIWelcomeMessage(ctx *sqls.TxContext, conversatio MessageType: enums.IMMessageTypeText, Content: content, Payload: payload, - SeqNo: repositories.MessageRepository.NextSeqNo(ctx.Tx, conversation.ID), SendStatus: enums.IMMessageStatusSent, SentAt: &now, AuditFields: models.AuditFields{ @@ -322,11 +321,11 @@ func (s *messageService) createAIWelcomeMessage(ctx *sqls.TxContext, conversatio return nil, err } agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, conversation.ID) - agentUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(agentReadState), enums.IMSenderTypeCustomer) + agentUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readMessageID(agentReadState), enums.IMSenderTypeCustomer) if err != nil { return nil, err } - customerUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(customerReadState), enums.IMSenderTypeAgent, enums.IMSenderTypeAI) + customerUnreadCount, err := ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readMessageID(customerReadState), enums.IMSenderTypeAgent, enums.IMSenderTypeAI) if err != nil { return nil, err } @@ -423,7 +422,6 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation, traceID = tracex.NormalizeRequestID(requestID) auditUserID = int64(0) auditUserName = "" - nextSeq = repositories.MessageRepository.NextSeqNo(sqls.DB(), conversation.ID) ) if operator != nil { auditUserID = operator.UserID @@ -443,7 +441,6 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation, MessageType: messageType, Content: content, Payload: payload, - SeqNo: nextSeq, SendStatus: enums.IMMessageStatusSent, SentAt: &now, AuditFields: models.AuditFields{ @@ -569,10 +566,10 @@ func (s *messageService) handleReadState(ctx *sqls.TxContext, senderType enums.I } } agentReadState, customerReadState := ConversationReadStateService.getConversationReadStates(ctx.Tx, conversation.ID) - if agentUnreadCount, err = ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readSeqNo(agentReadState), enums.IMSenderTypeCustomer); err != nil { + if agentUnreadCount, err = ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readMessageID(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 { + if customerUnreadCount, err = ConversationReadStateService.CountUnreadMessages(ctx, conversation.ID, s.readMessageID(customerReadState), enums.IMSenderTypeAgent, enums.IMSenderTypeAI); err != nil { return 0, 0, err } return agentUnreadCount, customerUnreadCount, nil @@ -706,9 +703,9 @@ func (s *messageService) suffixFilenameForSummary(filename string) string { return " " + filename } -func (s *messageService) readSeqNo(state *models.ConversationReadState) int64 { +func (s *messageService) readMessageID(state *models.ConversationReadState) int64 { if state == nil { return 0 } - return state.LastReadSeqNo + return state.LastReadMessageID } diff --git a/internal/services/message_service_test.go b/internal/services/message_service_test.go index 6ff8fc9..222e636 100644 --- a/internal/services/message_service_test.go +++ b/internal/services/message_service_test.go @@ -1,7 +1,9 @@ package services import ( + "fmt" "strings" + "sync" "testing" "time" @@ -163,9 +165,6 @@ func TestConversationCreateCreatesAIWelcomeMessage(t *testing.T) { if message.Content != "您好，请问有什么可以帮您？" { t.Fatalf("expected trimmed welcome content, got %q", message.Content) } - if message.SeqNo != 1 { - t.Fatalf("expected seq no 1, got %d", message.SeqNo) - } if message.SendStatus != enums.IMMessageStatusSent { t.Fatalf("expected sent status, got %d", message.SendStatus) } @@ -222,6 +221,134 @@ func TestSendCustomerMessageStoresRequestIDOnMessageAndEvent(t *testing.T) { } } +func TestSendCustomerMessagesConcurrentlyAssignsUniqueIDs(t *testing.T) { + db := setupMessageWelcomeTestDB(t) + aiAgent := createWelcomeTestAIAgent(t, db, "") + external := welcomeTestExternalUser("concurrent-user") + conversation, err := ConversationService.Create(external, 11, aiAgent.ID) + if err != nil { + t.Fatalf("create conversation: %v", err) + } + + const messageCount = 10 + var wg sync.WaitGroup + errCh := make(chan error, messageCount) + for i := 0; i < messageCount; i++ { + i := i + wg.Add(1) + go func() { + defer wg.Done() + _, err := MessageService.SendCustomerMessageWithRequestID( + conversation.ID, + fmt.Sprintf("client-msg-concurrent-%d", i), + enums.IMMessageTypeText, + "hello concurrent", + "", + external, + "trace-concurrent", + ) + errCh <- err + }() + } + wg.Wait() + close(errCh) + for err := range errCh { + if err != nil { + t.Fatalf("SendCustomerMessageWithRequestID() concurrent error = %v", err) + } + } + + var messages []models.Message + if err := db. + Where("conversation_id = ? AND sender_type = ?", conversation.ID, enums.IMSenderTypeCustomer). + Order("id ASC"). + Find(&messages).Error; err != nil { + t.Fatalf("find messages: %v", err) + } + if len(messages) != messageCount { + t.Fatalf("expected %d customer messages, got %d", messageCount, len(messages)) + } + seen := make(map[int64]struct{}, messageCount) + for _, message := range messages { + if message.ID <= 0 { + t.Fatalf("expected persisted message id, got %d", message.ID) + } + if _, ok := seen[message.ID]; ok { + t.Fatalf("duplicate message id %d", message.ID) + } + seen[message.ID] = struct{}{} + } +} + +func TestUnreadCountUsesLastReadMessageID(t *testing.T) { + db := setupMessageWelcomeTestDB(t) + aiAgent := createWelcomeTestAIAgent(t, db, "") + conversation := createMessageTestConversation(t, db, aiAgent.ID) + now := time.Now() + + messages := []models.Message{ + { + ConversationID: conversation.ID, + ClientMsgID: "read-message", + SenderType: enums.IMSenderTypeCustomer, + MessageType: enums.IMMessageTypeText, + Content: "read", + SendStatus: enums.IMMessageStatusSent, + SentAt: &now, + AuditFields: models.AuditFields{CreatedAt: now, UpdatedAt: now}, + }, + { + ConversationID: conversation.ID, + ClientMsgID: "unread-message-1", + SenderType: enums.IMSenderTypeCustomer, + MessageType: enums.IMMessageTypeText, + Content: "unread 1", + SendStatus: enums.IMMessageStatusSent, + SentAt: &now, + AuditFields: models.AuditFields{CreatedAt: now, UpdatedAt: now}, + }, + { + ConversationID: conversation.ID, + ClientMsgID: "unread-message-2", + SenderType: enums.IMSenderTypeCustomer, + MessageType: enums.IMMessageTypeText, + Content: "unread 2", + SendStatus: enums.IMMessageStatusSent, + SentAt: &now, + AuditFields: models.AuditFields{CreatedAt: now, UpdatedAt: now}, + }, + } + if err := db.Create(&messages).Error; err != nil { + t.Fatalf("create messages: %v", err) + } + + readState := &models.ConversationReadState{ + ConversationID: conversation.ID, + ReaderType: enums.IMSenderTypeAgent, + ReaderID: 1, + LastReadMessageID: messages[0].ID, + LastReadAt: &now, + AuditFields: models.AuditFields{CreatedAt: now, UpdatedAt: now}, + } + if err := db.Create(readState).Error; err != nil { + t.Fatalf("create read state: %v", err) + } + + err := sqls.WithTransaction(func(ctx *sqls.TxContext) error { + count, err := ConversationService.countUnreadByState(ctx, conversation.ID, readState, enums.IMSenderTypeCustomer) + if err != nil { + return err + } + if count != 2 { + t.Fatalf("unread count=%d want 2", count) + } + return nil + }) + if err != nil { + t.Fatalf("count unread: %v", err) + } +} + func TestSendAIMessageStoresWorkflowRunID(t *testing.T) { db := setupMessageWelcomeTestDB(t) aiAgent := createWelcomeTestAIAgent(t, db, "") diff --git a/internal/services/ws_realtime_types.go b/internal/services/ws_realtime_types.go index 504395d..1481384 100644 --- a/internal/services/ws_realtime_types.go +++ b/internal/services/ws_realtime_types.go @@ -146,7 +146,6 @@ type RealtimeMessageCreatedPayload struct { MessageType enums.IMMessageType `json:"messageType,omitempty"` Content string `json:"content,omitempty"` Payload string `json:"payload,omitempty"` - SeqNo int64 `json:"seqNo,omitempty"` SendStatus enums.IMMessageStatus `json:"sendStatus,omitempty"` SentAt string `json:"sentAt,omitempty"` } @@ -201,10 +200,8 @@ type RealtimeConversationChangedPayload struct { CustomerUnreadCount int `json:"customerUnreadCount,omitempty"` AgentUnreadCount int `json:"agentUnreadCount,omitempty"` CustomerLastReadMessageID int64 `json:"customerLastReadMessageId,omitempty"` - CustomerLastReadSeqNo int64 `json:"customerLastReadSeqNo,omitempty"` CustomerLastReadAt string `json:"customerLastReadAt,omitempty"` AgentLastReadMessageID int64 `json:"agentLastReadMessageId,omitempty"` - AgentLastReadSeqNo int64 `json:"agentLastReadSeqNo,omitempty"` AgentLastReadAt string `json:"agentLastReadAt,omitempty"` } diff --git a/internal/services/ws_service.go b/internal/services/ws_service.go index 66ba912..2f37acc 100644 --- a/internal/services/ws_service.go +++ b/internal/services/ws_service.go @@ -253,7 +253,9 @@ func (s *wsService) closeSession(session *ClientSession) { remaining := s.manager.Unregister(session) close(session.Send) - _ = session.Conn.Close() + if session.Conn != nil { + _ = session.Conn.Close() + } var discUserID int64 var discExternalID string @@ -309,7 +311,6 @@ func (s *wsService) PublishMessageCreated(conversation *models.Conversation, mes MessageType: message.MessageType, Content: content, Payload: payload, - SeqNo: message.SeqNo, SendStatus: message.SendStatus, SentAt: formatWsTime(message.SentAt), }, @@ -334,7 +335,6 @@ func (s *wsService) buildRealtimeMessage(item *models.Message) response.MessageR MessageType: item.MessageType, Content: content, Payload: payload, - SeqNo: item.SeqNo, SendStatus: item.SendStatus, SentAt: utils.FormatTimePtr(item.SentAt), DeliveredAt: utils.FormatTimePtr(item.DeliveredAt), @@ -389,7 +389,7 @@ func (s *wsService) fillRealtimeMessageUserName(ret *response.MessageResponse, u } func isRealtimeMessageRead(item *models.Message, state *models.ConversationReadState) bool { - return item != nil && state != nil && state.LastReadSeqNo >= item.SeqNo + return item != nil && state != nil && state.LastReadMessageID >= item.ID } func realtimeReadMessageAt(item *models.Message, state *models.ConversationReadState) string { @@ -438,10 +438,8 @@ func (s *wsService) PublishConversationChanged(conversation *models.Conversation CustomerUnreadCount: conversation.CustomerUnreadCount, AgentUnreadCount: conversation.AgentUnreadCount, CustomerLastReadMessageID: readStateMessageID(customerReadState), - CustomerLastReadSeqNo: readStateSeqNo(customerReadState), CustomerLastReadAt: readStateAt(customerReadState), AgentLastReadMessageID: readStateMessageID(agentReadState), - AgentLastReadSeqNo: readStateSeqNo(agentReadState), AgentLastReadAt: readStateAt(agentReadState), }, }) @@ -478,13 +476,6 @@ func readStateMessageID(state *models.ConversationReadState) int64 { return state.LastReadMessageID } -func readStateSeqNo(state *models.ConversationReadState) int64 { - if state == nil { - return 0 - } - return state.LastReadSeqNo -} - func readStateAt(state *models.ConversationReadState) string { if state == nil { return ""