From 3d9817c9f542ceb48230d2c2a960c0b0b60ecee8 Mon Sep 17 00:00:00 2001 From: Rogee Date: Thu, 20 Aug 2026 14:30:05 +0800 Subject: [PATCH] H-337: fix Web Channel availability and realtime delivery (#60) * H-337: fix Web Channel availability and realtime delivery * fix(widget): preserve realtime sender and activity contracts * fix(widget): keep realtime sender payloads consistent * fix(widget): make public message persistence atomic * fix(inbox): keep availability projection out of schema --------- Co-authored-by: Rogee --- backend/internal/app/bootstrap.go | 2 +- .../api/v1/inbox_member_handler_test.go | 2 +- .../internal/handler/widget/coverage3_test.go | 15 - .../handler/widget/coverage3b_test.go | 31 -- .../internal/handler/widget/coverage4_test.go | 11 - .../internal/handler/widget/widget_handler.go | 49 +-- .../handler/widget/widget_handler_test.go | 74 +---- backend/internal/model/inbox_member.go | 11 +- .../internal/repository/inbox_member_repo.go | 27 +- .../repository/inbox_member_repo_test.go | 7 +- .../service/captain_conversation_service.go | 51 ++-- backend/internal/service/message_service.go | 50 ++++ .../internal/service/message_service_test.go | 91 ++++++ backend/internal/service/widget_service.go | 151 +++++++--- .../internal/service/widget_service_test.go | 281 ++++++++++++++++++ backend/internal/wsevent/bridge_listener.go | 77 ++++- .../internal/wsevent/bridge_listener_test.go | 86 +++++- 17 files changed, 769 insertions(+), 247 deletions(-) diff --git a/backend/internal/app/bootstrap.go b/backend/internal/app/bootstrap.go index 9c73eb28..467c7a8e 100644 --- a/backend/internal/app/bootstrap.go +++ b/backend/internal/app/bootstrap.go @@ -775,7 +775,7 @@ func Bootstrap(env string) (*App, error) { widgetService := service.NewWidgetService(inboxRepo, contactRepo, contactInboxRepo, conversationRepo, messageRepo, widgetTypingAdapter, widgetThemeConfigRepo, preChatFormRepo, widgetFileUploadRepo, widgetOfflineMessageRepo, inboxMemberRepo, tagRepo, campaignRepo) widgetService.SetWorkerPool(workerPool) widgetService.SetDispatcher(channelDispatcher) - widgetHandler := widget.NewHandler(widgetService).WithEventPublisher(eventPublisher) + widgetHandler := widget.NewHandler(widgetService) // Upload: DirectUpload repo + service + handler (account-level + widget direct uploads) directUploadRepo := repository.NewDirectUploadRepo(db) diff --git a/backend/internal/handler/api/v1/inbox_member_handler_test.go b/backend/internal/handler/api/v1/inbox_member_handler_test.go index b197c8c9..099519a9 100644 --- a/backend/internal/handler/api/v1/inbox_member_handler_test.go +++ b/backend/internal/handler/api/v1/inbox_member_handler_test.go @@ -35,7 +35,7 @@ func (s *InboxMemberHandlerTestSuite) SetupSuite() { Logger: logger.Default.LogMode(logger.Silent), }) s.Require().NoError(err) - s.Require().NoError(db.AutoMigrate(&model.Account{}, &model.Inbox{}, &model.User{}, &model.InboxMember{})) + s.Require().NoError(db.AutoMigrate(&model.Account{}, &model.Inbox{}, &model.User{}, &model.AccountUser{}, &model.InboxMember{})) s.db = db repo := repository.NewInboxMemberRepo(db) diff --git a/backend/internal/handler/widget/coverage3_test.go b/backend/internal/handler/widget/coverage3_test.go index 91d68e6a..5467a0b3 100644 --- a/backend/internal/handler/widget/coverage3_test.go +++ b/backend/internal/handler/widget/coverage3_test.go @@ -224,21 +224,6 @@ func TestWidgetHandler_Cov3_PublicGetContact_InvalidInbox_Cov3(t *testing.T) { assert.True(t, w.Code == http.StatusNotFound || w.Code == http.StatusBadRequest, "expected 404 or 400, got %d: %s", w.Code, w.Body.String()) } -// ---------- uintFromPayload unit test ---------- - -func TestWidgetHandler_Cov3_UintFromPayload_Cov3(t *testing.T) { - assert.Equal(t, uint(42), uintFromPayload(uint(42))) - assert.Equal(t, uint(42), uintFromPayload(int(42))) - assert.Equal(t, uint(0), uintFromPayload(int(-1))) - assert.Equal(t, uint(42), uintFromPayload(int64(42))) - assert.Equal(t, uint(0), uintFromPayload(int64(-1))) - assert.Equal(t, uint(42), uintFromPayload(float64(42))) - assert.Equal(t, uint(0), uintFromPayload(float64(-1))) - assert.Equal(t, uint(0), uintFromPayload("not a number")) - assert.Equal(t, uint(0), uintFromPayload(nil)) - assert.Equal(t, uint(0), uintFromPayload(uint(0))) -} - // ---------- GetCableToken ---------- func TestWidgetHandler_Cov3_GetCableToken_NoToken_Cov3(t *testing.T) { diff --git a/backend/internal/handler/widget/coverage3b_test.go b/backend/internal/handler/widget/coverage3b_test.go index d3d344bb..18871340 100644 --- a/backend/internal/handler/widget/coverage3b_test.go +++ b/backend/internal/handler/widget/coverage3b_test.go @@ -311,37 +311,6 @@ func TestWidgetHandler_Cov3_GetConversations_ChatwootRoute_Cov3(t *testing.T) { } } -// ---------- SendMessage with event publisher ---------- - -func TestWidgetHandler_Cov3_SendMessage_WithEventPublisher_Cov3(t *testing.T) { - db, router, handler := setupWidgetHandlerTest(t) - seedWidgetHandlerData(t, db) - - // Set event publisher - publisher := &recordingWidgetEventPublisher{} - handler.WithEventPublisher(publisher) - - // Config to get a widget token - wConfig := httptest.NewRecorder() - reqConfig, _ := http.NewRequest("POST", "/api/v1/widget/config?website_token=handler_ws_token_123", nil) - router.ServeHTTP(wConfig, reqConfig) - require.Equal(t, http.StatusOK, wConfig.Code) - var configResp map[string]interface{} - require.NoError(t, json.Unmarshal(wConfig.Body.Bytes(), &configResp)) - authToken := configResp["contact"].(map[string]interface{})["pubsub_token"].(string) - - // Send message - messageBody, _ := json.Marshal(map[string]interface{}{"message": map[string]interface{}{"content": "hello with publisher"}}) - wMsg := httptest.NewRecorder() - reqMsg, _ := http.NewRequest("POST", "/api/v1/widget/messages?cw_conversation="+authToken, bytes.NewReader(messageBody)) - reqMsg.Header.Set("Content-Type", "application/json") - router.ServeHTTP(wMsg, reqMsg) - require.Equal(t, http.StatusOK, wMsg.Code, wMsg.Body.String()) - - // The event publisher should have been called - assert.NotZero(t, publisher.accountID) -} - // ---------- SendTranscript with valid token (needs conversation) ---------- func TestWidgetHandler_Cov3_SendTranscript_WithConversation_Cov3(t *testing.T) { diff --git a/backend/internal/handler/widget/coverage4_test.go b/backend/internal/handler/widget/coverage4_test.go index d860e71e..8f91993b 100644 --- a/backend/internal/handler/widget/coverage4_test.go +++ b/backend/internal/handler/widget/coverage4_test.go @@ -86,17 +86,6 @@ func TestConfig_InvalidWebsiteToken_Cov4(t *testing.T) { assert.Equal(t, http.StatusNotFound, w.Code) } -func TestConfig_WithEventPublisher_Cov4(t *testing.T) { - db, router, handler := setupWidgetHandlerTest(t) - seedWidgetHandlerData(t, db) - pub := &recordingWidgetEventPublisher{} - handler.WithEventPublisher(pub) - w := httptest.NewRecorder() - req, _ := http.NewRequest("POST", "/api/v1/widget/config?website_token=handler_ws_token_123", nil) - router.ServeHTTP(w, req) - assert.Equal(t, http.StatusOK, w.Code) -} - // ============================================================ // SendMessage — edge cases // ============================================================ diff --git a/backend/internal/handler/widget/widget_handler.go b/backend/internal/handler/widget/widget_handler.go index 659a4149..6a56c30b 100644 --- a/backend/internal/handler/widget/widget_handler.go +++ b/backend/internal/handler/widget/widget_handler.go @@ -12,20 +12,14 @@ import ( "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/service" "github.com/gochat/gochat/internal/webhookutil" - wspkg "github.com/gochat/gochat/internal/ws" ) -type widgetEventPublisher interface { - PublishWidgetEvent(accountID uint, pubsubToken string, eventType string, payload interface{}) -} - // WidgetHandler handles the public-facing widget API endpoints. // These endpoints are accessed by the embedded JS widget on customer websites // and do not require agent JWT authentication — they use a widget_token instead. // Reference: Chatwoot app/controllers/api/v1/widget_messages_controller.rb type WidgetHandler struct { - widgetService *service.WidgetService - eventPublisher widgetEventPublisher + widgetService *service.WidgetService } // NewHandler creates a new WidgetHandler with the WidgetService dependency. @@ -35,11 +29,6 @@ func NewHandler(widgetService *service.WidgetService) *WidgetHandler { } } -func (h *WidgetHandler) WithEventPublisher(publisher widgetEventPublisher) *WidgetHandler { - h.eventPublisher = publisher - return h -} - // Init handles widget initialization — authenticates/creates a contact // and returns a widget_token (pubsub_token) for subsequent requests. // POST /widget/init @@ -212,7 +201,6 @@ func (h *WidgetHandler) SendMessage(c *gin.Context) { if len(resp.Attachments) > 0 { payload["attachments"] = widgetAttachmentPayloads(resp.Attachments) } - h.publishWidgetMessageEvent(req.WidgetToken, payload) c.JSON(http.StatusOK, resp) return } @@ -220,7 +208,6 @@ func (h *WidgetHandler) SendMessage(c *gin.Context) { if len(resp.Attachments) > 0 { payload["attachments"] = widgetAttachmentPayloads(resp.Attachments) } - h.publishWidgetMessageEvent(req.WidgetToken, payload) c.JSON(http.StatusOK, payload) } @@ -262,40 +249,6 @@ func (h *WidgetHandler) UpdateMessage(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"contact": widgetContactFullPayload(contact)}) } -func (h *WidgetHandler) publishWidgetMessageEvent(pubsubToken string, payload gin.H) { - if h.eventPublisher == nil || payload == nil { - return - } - accountID := uintFromPayload(payload["account_id"]) - if accountID == 0 { - accountID = uintFromPayload(payload["accountId"]) - } - if accountID == 0 { - return - } - h.eventPublisher.PublishWidgetEvent(accountID, pubsubToken, wspkg.EventMessageCreated, payload) -} - -func uintFromPayload(value any) uint { - switch typed := value.(type) { - case uint: - return typed - case int: - if typed > 0 { - return uint(typed) - } - case int64: - if typed > 0 { - return uint(typed) - } - case float64: - if typed > 0 { - return uint(typed) - } - } - return 0 -} - // GetLatestMessages implements Chatwoot's GET /api/v1/widget/messages endpoint. func (h *WidgetHandler) GetLatestMessages(c *gin.Context) { widgetToken := widgetTokenFromRequest(c) diff --git a/backend/internal/handler/widget/widget_handler_test.go b/backend/internal/handler/widget/widget_handler_test.go index f854a187..d6192350 100644 --- a/backend/internal/handler/widget/widget_handler_test.go +++ b/backend/internal/handler/widget/widget_handler_test.go @@ -43,13 +43,6 @@ func (n *noopTypingIndicatorWidget) SetTypingOff(_ context.Context, _ uint, _ ui return nil } -type recordingWidgetEventPublisher struct { - accountID uint - pubsubToken string - eventType string - payload interface{} -} - func TestPublicMessagePayloadSanitizesIdentity(t *testing.T) { senderID := uint(9) message := model.Message{ @@ -73,13 +66,6 @@ func TestPublicMessagePayloadSanitizesIdentity(t *testing.T) { assert.Empty(t, publicMessagePayload(message, model.Conversation{})["content_attributes"]) } -func (p *recordingWidgetEventPublisher) PublishWidgetEvent(accountID uint, pubsubToken string, eventType string, payload interface{}) { - p.accountID = accountID - p.pubsubToken = pubsubToken - p.eventType = eventType - p.payload = payload -} - type recordingWidgetTranscriptDeliverer struct { requests []automation.AutomationTranscriptRequest err error @@ -779,42 +765,6 @@ func TestWidgetHandler_ChatwootMessages_AuthTokenAndNestedPayload(t *testing.T) assertWidgetContactFixtureShape(t, contactResp) } -func TestWidgetHandler_ChatwootMessagePublishesWidgetRealtimePayload(t *testing.T) { - db, router, handler := setupWidgetHandlerTest(t) - _, _ = seedWidgetHandlerData(t, db) - publisher := &recordingWidgetEventPublisher{} - handler.WithEventPublisher(publisher) - - wConfig := httptest.NewRecorder() - reqConfig, _ := http.NewRequest("POST", "/api/v1/widget/config?website_token=handler_ws_token_123", nil) - router.ServeHTTP(wConfig, reqConfig) - require.Equal(t, http.StatusOK, wConfig.Code) - - var configResp map[string]interface{} - require.NoError(t, json.Unmarshal(wConfig.Body.Bytes(), &configResp)) - authToken := configResp["contact"].(map[string]interface{})["pubsub_token"].(string) - - body, _ := json.Marshal(map[string]interface{}{ - "message": map[string]interface{}{"content": "Realtime widget message"}, - }) - wMessage := httptest.NewRecorder() - reqMessage, _ := http.NewRequest("POST", "/api/v1/widget/messages", bytes.NewReader(body)) - reqMessage.Header.Set("Content-Type", "application/json") - reqMessage.Header.Set("X-Auth-Token", authToken) - router.ServeHTTP(wMessage, reqMessage) - require.Equal(t, http.StatusOK, wMessage.Code) - - assert.Equal(t, uint(1), publisher.accountID) - assert.Equal(t, authToken, publisher.pubsubToken) - assert.Equal(t, ws.EventMessageCreated, publisher.eventType) - payload, ok := publisher.payload.(gin.H) - require.True(t, ok, "expected widget realtime payload to use Chatwoot widget message shape") - assert.Equal(t, "Realtime widget message", payload["content"]) - assert.Equal(t, uint(1), payload["account_id"]) - assert.NotEmpty(t, payload["conversation_id"]) - assertWidgetMessageFixtureShape(t, payload) -} - func TestWidgetHandler_ChatwootMessagesIndexFiltersInternalMessages(t *testing.T) { db, router, _ := setupWidgetHandlerTest(t) _, inbox := seedWidgetHandlerData(t, db) @@ -1656,21 +1606,26 @@ func TestWidgetHandler_ChatwootInboxMembers_Success(t *testing.T) { db, router, _ := setupWidgetHandlerTest(t) _, inbox := seedWidgetHandlerData(t, db) - user := &model.User{ + userA := &model.User{ AccountID: inbox.AccountID, - Name: "Ada Agent", + Name: "Agent A", Email: "ada@handler.test", Password: "secret", AvatarURL: "https://example.test/ada.png", Active: true, } - require.NoError(t, db.Create(user).Error) + userB := &model.User{AccountID: inbox.AccountID, Name: "Agent B", Email: "b@handler.test", Password: "secret", Active: true} + require.NoError(t, db.Create(userA).Error) + require.NoError(t, db.Create(userB).Error) + require.NoError(t, db.Create(&model.AccountUser{AccountID: inbox.AccountID, UserID: userA.ID, Role: "agent", Availability: "online"}).Error) + require.NoError(t, db.Create(&model.AccountUser{AccountID: inbox.AccountID, UserID: userB.ID, Role: "agent", Availability: "online"}).Error) require.NoError(t, db.Create(&model.InboxMember{ InboxID: inbox.ID, - UserID: user.ID, + UserID: userA.ID, Role: "agent", - AvailabilityStatus: "online", + AvailabilityStatus: "offline", }).Error) + require.NoError(t, db.Create(&model.InboxMember{InboxID: inbox.ID, UserID: userB.ID, Role: "agent", AvailabilityStatus: "offline"}).Error) w := httptest.NewRecorder() req, _ := http.NewRequest("GET", "/api/v1/widget/inbox_members?website_token=handler_ws_token_123", nil) @@ -1680,10 +1635,11 @@ func TestWidgetHandler_ChatwootInboxMembers_Success(t *testing.T) { var resp map[string]interface{} require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) payload := resp["payload"].([]interface{}) - require.Len(t, payload, 1) - member := payload[0].(map[string]interface{}) - assert.Equal(t, "Ada Agent", member["name"]) - assert.Equal(t, "online", member["availability_status"]) + require.Len(t, payload, 2) + assert.Equal(t, "Agent A", payload[0].(map[string]interface{})["name"]) + assert.Equal(t, "online", payload[0].(map[string]interface{})["availability_status"]) + assert.Equal(t, "Agent B", payload[1].(map[string]interface{})["name"]) + assert.Equal(t, "online", payload[1].(map[string]interface{})["availability_status"]) } func TestWidgetHandler_ChatwootInboxMembers_InvalidWebsiteTokenReturnsNotFound(t *testing.T) { diff --git a/backend/internal/model/inbox_member.go b/backend/internal/model/inbox_member.go index 39b6ef44..e8c099ac 100644 --- a/backend/internal/model/inbox_member.go +++ b/backend/internal/model/inbox_member.go @@ -4,13 +4,14 @@ package model // Reference: Chatwoot InboxMember model + P2B M2 spec type InboxMember struct { Base - InboxID uint `gorm:"column:inbox_id;not null;index" json:"inbox_id"` - UserID uint `gorm:"column:user_id;not null;index" json:"user_id"` - Role string `gorm:"column:role;size:50;default:agent" json:"role"` // agent, supervisor - AvailabilityStatus string `gorm:"column:availability_status;size:50;default:offline" json:"availability_status"` // online, offline, busy + InboxID uint `gorm:"column:inbox_id;not null;index" json:"inbox_id"` + UserID uint `gorm:"column:user_id;not null;index" json:"user_id"` + Role string `gorm:"column:role;size:50;default:agent" json:"role"` // agent, supervisor + AvailabilityStatus string `gorm:"column:availability_status;size:50;default:offline" json:"availability_status"` // online, offline, busy + AccountAvailability string `gorm:"-" json:"-"` Inbox Inbox `gorm:"foreignKey:InboxID" json:"inbox,omitempty"` User User `gorm:"foreignKey:UserID" json:"user,omitempty"` } -func (InboxMember) TableName() string { return "inbox_members" } \ No newline at end of file +func (InboxMember) TableName() string { return "inbox_members" } diff --git a/backend/internal/repository/inbox_member_repo.go b/backend/internal/repository/inbox_member_repo.go index 41b20f13..ce4fd96f 100644 --- a/backend/internal/repository/inbox_member_repo.go +++ b/backend/internal/repository/inbox_member_repo.go @@ -41,10 +41,29 @@ func (r *InboxMemberRepo) FindByInboxAndUser(ctx context.Context, inboxID, userI // FindByInbox retrieves all members (agents) assigned to an inbox. func (r *InboxMemberRepo) FindByInbox(ctx context.Context, inboxID uint) ([]model.InboxMember, error) { - var ims []model.InboxMember - err := r.withAgentPreloads(r.db.WithContext(ctx)).Where("inbox_id = ?", inboxID). - Order("id ASC").Find(&ims).Error - return ims, err + type inboxMemberWithAvailability struct { + model.InboxMember `gorm:"embedded"` + AccountAvailability string `gorm:"column:account_availability"` + } + + var rows []inboxMemberWithAvailability + err := r.withAgentPreloads(r.db.WithContext(ctx)). + Model(&model.InboxMember{}). + Select("inbox_members.*, COALESCE(account_users.availability, 'offline') AS account_availability"). + Joins("LEFT JOIN inboxes ON inboxes.id = inbox_members.inbox_id"). + Joins("LEFT JOIN account_users ON account_users.account_id = inboxes.account_id AND account_users.user_id = inbox_members.user_id AND account_users.deleted_at IS NULL"). + Where("inbox_members.inbox_id = ?", inboxID). + Order("inbox_members.id ASC").Find(&rows).Error + if err != nil { + return nil, err + } + + ims := make([]model.InboxMember, len(rows)) + for i := range rows { + ims[i] = rows[i].InboxMember + ims[i].AccountAvailability = rows[i].AccountAvailability + } + return ims, nil } // FindByUser retrieves all inboxes a user is assigned to. diff --git a/backend/internal/repository/inbox_member_repo_test.go b/backend/internal/repository/inbox_member_repo_test.go index 4f75bac2..529cdc70 100644 --- a/backend/internal/repository/inbox_member_repo_test.go +++ b/backend/internal/repository/inbox_member_repo_test.go @@ -103,11 +103,12 @@ func TestInboxMemberRepo_FindByInbox(t *testing.T) { db := setupTestDB(t, &model.InboxMember{}) repo := NewInboxMemberRepo(db) - _, user1, inbox := createInboxMemberPrereqs(t, db) + account, user1, inbox := createInboxMemberPrereqs(t, db) // Create a second user user2 := &model.User{Name: "IMTestUser2", Email: "im2@example.com", Provider: "email"} require.NoError(t, db.Create(user2).Error) + require.NoError(t, db.Create(&model.AccountUser{AccountID: account.ID, UserID: user1.ID, Role: "agent", Availability: "online"}).Error) im1 := &model.InboxMember{InboxID: inbox.ID, UserID: user1.ID, Role: "agent"} require.NoError(t, db.Create(im1).Error) @@ -121,6 +122,8 @@ func TestInboxMemberRepo_FindByInbox(t *testing.T) { // Verify User is preloaded assert.Equal(t, "IMTestUser", members[0].User.Name) assert.Equal(t, "IMTestUser2", members[1].User.Name) + assert.Equal(t, "online", members[0].AccountAvailability) + assert.Equal(t, "offline", members[1].AccountAvailability) // Verify ordering by ID ASC assert.Equal(t, im1.ID, members[0].ID) @@ -240,4 +243,4 @@ func TestInboxMemberRepo_DeleteByInbox(t *testing.T) { members, err := repo.FindByInbox(context.Background(), inbox.ID) assert.NoError(t, err) assert.Len(t, members, 0) -} \ No newline at end of file +} diff --git a/backend/internal/service/captain_conversation_service.go b/backend/internal/service/captain_conversation_service.go index db181a0b..fda803c9 100644 --- a/backend/internal/service/captain_conversation_service.go +++ b/backend/internal/service/captain_conversation_service.go @@ -393,63 +393,74 @@ func RegisterCaptainConversationJobs(wp *worker.WorkerPool, svc *CaptainConversa } func EnqueueCaptainConversationResponseForMessage(ctx context.Context, wp *worker.WorkerPool, db *gorm.DB, messageID uint) (*model.BackgroundJob, error) { - if wp == nil || db == nil || messageID == 0 || !db.Migrator().HasTable(&model.CaptainInbox{}) { - return nil, nil + job, created, err := enqueueCaptainConversationResponseForMessageInTransaction(ctx, wp, db, messageID) + if err != nil { + return nil, err + } + if created { + wp.Publish(ctx, job) + } + return job, nil +} + +func enqueueCaptainConversationResponseForMessageInTransaction(ctx context.Context, wp *worker.WorkerPool, tx *gorm.DB, messageID uint) (*model.BackgroundJob, bool, error) { + if wp == nil || tx == nil || messageID == 0 || !tx.Migrator().HasTable(&model.CaptainInbox{}) { + return nil, false, nil } var message model.Message - if err := db.WithContext(ctx).First(&message, messageID).Error; err != nil { - return nil, err + if err := tx.WithContext(ctx).First(&message, messageID).Error; err != nil { + return nil, false, err } if message.MessageType != string(model.MessageTypeIncoming) || message.Private { - return nil, nil + return nil, false, nil } var conversation model.Conversation - if err := db.WithContext(ctx).Where("account_id = ? AND id = ?", message.AccountID, message.ConversationID).First(&conversation).Error; err != nil { - return nil, err + if err := tx.WithContext(ctx).Where("account_id = ? AND id = ?", message.AccountID, message.ConversationID).First(&conversation).Error; err != nil { + return nil, false, err } if conversation.Status != string(model.ConversationStatusPending) { - return nil, nil + return nil, false, nil } var assistantID uint var takeoverVersion uint if conversation.AssigneeAgentBotID != nil { var bot model.AgentBot - if err := db.WithContext(ctx).Where("id = ? AND bot_type = ?", *conversation.AssigneeAgentBotID, "captain").First(&bot).Error; err != nil { + if err := tx.WithContext(ctx).Where("id = ? AND bot_type = ?", *conversation.AssigneeAgentBotID, "captain").First(&bot).Error; err != nil { if err == gorm.ErrRecordNotFound { - return nil, nil + return nil, false, nil } - return nil, err + return nil, false, err } assistantID = extractAssistantIDFromBotConfig(bot.Config) takeoverVersion = conversation.AITakeoverVersion } else { var preference model.CaptainPreference - if err := db.WithContext(ctx).Where("account_id = ? AND auto_reply_enabled = ?", conversation.AccountID, true).First(&preference).Error; err != nil { + if err := tx.WithContext(ctx).Where("account_id = ? AND auto_reply_enabled = ?", conversation.AccountID, true).First(&preference).Error; err != nil { if err == gorm.ErrRecordNotFound { - return nil, nil + return nil, false, nil } - return nil, err + return nil, false, err } var ci model.CaptainInbox - if err := db.WithContext(ctx).Where("account_id = ? AND inbox_id = ?", conversation.AccountID, conversation.InboxID).First(&ci).Error; err != nil { + if err := tx.WithContext(ctx).Where("account_id = ? AND inbox_id = ?", conversation.AccountID, conversation.InboxID).First(&ci).Error; err != nil { if err == gorm.ErrRecordNotFound { - return nil, nil + return nil, false, nil } - return nil, err + return nil, false, err } assistantID = ci.AssistantID } if assistantID == 0 { - return nil, nil + return nil, false, nil } options := []worker.EnqueueOption{ worker.WithMaxAttempts(3), worker.WithIdempotencyKey(fmt.Sprintf("captain:conversation_response_builder:message:%d", message.ID)), } - if scheduledAt := captainConversationResponseScheduledAt(ctx, db, message.ID); !scheduledAt.IsZero() { + if scheduledAt := captainConversationResponseScheduledAt(ctx, tx, message.ID); !scheduledAt.IsZero() { options = append(options, worker.WithScheduledAt(scheduledAt)) } - return wp.Enqueue(ctx, TaskTypeCaptainConversationResponseBuilder, captainConversationResponseBuilderJob{AccountID: conversation.AccountID, ConversationID: conversation.ID, AssistantID: assistantID, MessageID: message.ID, TakeoverVersion: takeoverVersion}, options...) + return wp.EnqueueInTransaction(ctx, tx, TaskTypeCaptainConversationResponseBuilder, captainConversationResponseBuilderJob{AccountID: conversation.AccountID, ConversationID: conversation.ID, AssistantID: assistantID, MessageID: message.ID, TakeoverVersion: takeoverVersion}, options...) } func captainConversationResponseScheduledAt(ctx context.Context, db *gorm.DB, messageID uint) time.Time { diff --git a/backend/internal/service/message_service.go b/backend/internal/service/message_service.go index a4a84de1..97d36155 100644 --- a/backend/internal/service/message_service.go +++ b/backend/internal/service/message_service.go @@ -76,12 +76,62 @@ func (s *MessageService) dispatchMessageEvent(ctx context.Context, eventType cha event.UserID = *message.SenderID } event.Data["message"] = message + if eventType == channel.EventMessageCreated || eventType == channel.EventMessageUpdated || eventType == channel.EventMessageDeleted { + s.addWebWidgetEventContext(ctx, event, message) + } applogger.L().Infof("dispatching event %s for message %d", eventType, message.ID) if err := s.dispatcher.Dispatch(ctx, event); err != nil { applogger.L().Errorf("failed to dispatch event %s for message %d: %v", eventType, message.ID, err) } } +func (s *MessageService) addWebWidgetEventContext(ctx context.Context, event *channel.ChannelEvent, message *model.Message) { + var inbox model.Inbox + if err := s.repo.DB().WithContext(ctx).First(&inbox, message.InboxID).Error; err != nil || inbox.ChannelType != string(channel.ChannelWebWidget) { + return + } + event.Channel = channel.ChannelWebWidget + event.Data["inbox"] = &inbox + + var conversation model.Conversation + if err := s.repo.DB().WithContext(ctx).First(&conversation, message.ConversationID).Error; err != nil { + return + } + event.ContactID = conversation.ContactID + event.Data["conversation"] = &conversation + + var contact model.Contact + if err := s.repo.DB().WithContext(ctx).First(&contact, conversation.ContactID).Error; err == nil { + event.Data["contact"] = &contact + } + if message.SenderID != nil { + switch strings.ToLower(strings.TrimSpace(message.SenderType)) { + case "user": + var sender model.User + if err := s.repo.DB().WithContext(ctx).First(&sender, *message.SenderID).Error; err == nil { + event.Data["sender"] = &sender + } + case "agentbot", "agent_bot": + var sender model.AgentBot + if err := s.repo.DB().WithContext(ctx).First(&sender, *message.SenderID).Error; err == nil { + event.Data["sender"] = &sender + } + case "captain::assistant", "captainassistant", "captain_assistant": + var sender model.CaptainAssistant + if err := s.repo.DB().WithContext(ctx).First(&sender, *message.SenderID).Error; err == nil { + event.Data["sender"] = &sender + } + } + } + if message.Private || message.MessageType == string(model.MessageTypeActivity) || conversation.ContactInboxID == nil { + return + } + var contactInbox model.ContactInbox + if err := s.repo.DB().WithContext(ctx).Select("pubsub_token").First(&contactInbox, *conversation.ContactInboxID).Error; err == nil { + event.Data["widget_token"] = contactInbox.PubsubToken + } +} + // ListByConversation retrieves all messages for a conversation. func (s *MessageService) ListByConversation(ctx context.Context, conversationID uint, offset, limit int) ([]model.Message, int64, error) { return s.repo.FindByConversation(ctx, conversationID, offset, limit) diff --git a/backend/internal/service/message_service_test.go b/backend/internal/service/message_service_test.go index 47b55ffe..caa901f5 100644 --- a/backend/internal/service/message_service_test.go +++ b/backend/internal/service/message_service_test.go @@ -55,6 +55,17 @@ type mockRetryListener struct { lastData map[string]interface{} } +type messageEventListener struct { + events []*channel.ChannelEvent +} + +func (l *messageEventListener) Name() string { return "message-event-listener" } + +func (l *messageEventListener) OnEvent(_ context.Context, event *channel.ChannelEvent) error { + l.events = append(l.events, event) + return nil +} + func (l *mockRetryListener) Name() string { return "mock-retry-listener" } @@ -93,6 +104,86 @@ func setupMessageServiceWithDefaultLLM(t *testing.T) (*gorm.DB, *repository.Mess return setupMessageService(t, mockLLM) } +func TestMessageService_WebWidgetReplyCarriesRealtimeContext(t *testing.T) { + db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t) + account := createTestAccount(t, db) + agent := createTestUser(t, db, account.ID) + inbox := createTestInbox(t, db, account.ID, string(channel.ChannelWebWidget)) + contact := createTestContact(t, db, account.ID) + contactInbox := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, SourceID: "visitor", PubsubToken: "visitor-token"} + require.NoError(t, db.Create(contactInbox).Error) + conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID) + require.NoError(t, db.Model(conversation).Update("contact_inbox_id", contactInbox.ID).Error) + + listener := &messageEventListener{} + dispatcher.Register(listener) + _, err := svc.Create(context.Background(), account.ID, agent.ID, CreateMessageRequest{ + ConversationID: conversation.ID, + Content: "agent reply", + ContentType: "text", + MessageType: "outgoing", + }) + require.NoError(t, err) + + var created *channel.ChannelEvent + for _, event := range listener.events { + if event.Type == channel.EventMessageCreated { + created = event + break + } + } + require.NotNil(t, created) + assert.Equal(t, channel.ChannelWebWidget, created.Channel) + assert.Equal(t, "visitor-token", created.Data["widget_token"]) + assert.IsType(t, &model.Conversation{}, created.Data["conversation"]) + assert.IsType(t, &model.Contact{}, created.Data["contact"]) + assert.IsType(t, &model.User{}, created.Data["sender"]) +} + +func TestMessageService_WebWidgetReplyResolvesAutomatedSenders(t *testing.T) { + db, _, _, svc := setupMessageServiceWithDefaultLLM(t) + require.NoError(t, db.AutoMigrate(&model.AgentBot{}, &model.CaptainAssistant{})) + account := createTestAccount(t, db) + inbox := createTestInbox(t, db, account.ID, string(channel.ChannelWebWidget)) + contact := createTestContact(t, db, account.ID) + contactInbox := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, SourceID: "visitor", PubsubToken: "visitor-token"} + require.NoError(t, db.Create(contactInbox).Error) + conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID) + require.NoError(t, db.Model(conversation).Update("contact_inbox_id", contactInbox.ID).Error) + + bot := &model.AgentBot{AccountID: &account.ID, Name: "Reply Bot", BotType: "webhook"} + require.NoError(t, db.Create(bot).Error) + assistant := &model.CaptainAssistant{AccountID: account.ID, Name: "Captain", Config: json.RawMessage(`{}`), Status: model.AssistantStatusActive} + require.NoError(t, db.Create(assistant).Error) + + tests := []struct { + name string + senderType string + senderID uint + expected interface{} + }{ + {name: "agent bot", senderType: "AgentBot", senderID: bot.ID, expected: &model.AgentBot{}}, + {name: "captain namespaced", senderType: "Captain::Assistant", senderID: assistant.ID, expected: &model.CaptainAssistant{}}, + {name: "captain alias", senderType: "CaptainAssistant", senderID: assistant.ID, expected: &model.CaptainAssistant{}}, + {name: "unknown", senderType: "CustomSender", senderID: assistant.ID}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + event := channel.NewChannelEvent(channel.EventMessageCreated, channel.ChannelWebWidget, account.ID, inbox.ID) + message := &model.Message{ConversationID: conversation.ID, InboxID: inbox.ID, SenderID: &tt.senderID, SenderType: tt.senderType} + svc.addWebWidgetEventContext(context.Background(), event, message) + + assert.Equal(t, "visitor-token", event.Data["widget_token"]) + assert.IsType(t, &model.Contact{}, event.Data["contact"]) + if tt.expected == nil { + assert.NotContains(t, event.Data, "sender") + } else { + assert.IsType(t, tt.expected, event.Data["sender"]) + } + }) + } +} + // ========== ListByConversation 测试 ========== func TestMessageService_ListByConversation(t *testing.T) { diff --git a/backend/internal/service/widget_service.go b/backend/internal/service/widget_service.go index 33634439..d48262b6 100644 --- a/backend/internal/service/widget_service.go +++ b/backend/internal/service/widget_service.go @@ -366,6 +366,7 @@ func (s *WidgetService) SendMessage(ctx context.Context, req WidgetSendMessageRe // Resolve conversation var conversation *model.Conversation + conversationCreated := false if req.ConversationID != nil { conversation, err = s.conversationRepo.FindByID(ctx, *req.ConversationID) if err != nil { @@ -387,6 +388,7 @@ func (s *WidgetService) SendMessage(ctx context.Context, req WidgetSendMessageRe if err != nil { return nil, fmt.Errorf("failed to create conversation: %w", err) } + conversationCreated = true } } @@ -408,24 +410,38 @@ func (s *WidgetService) SendMessage(ctx context.Context, req WidgetSendMessageRe } } - if err := s.messageRepo.Create(ctx, &msg); err != nil { - return nil, fmt.Errorf("failed to create message: %w", err) - } - if err := s.reopenWidgetConversationForIncomingMessage(ctx, conversation); err != nil { - return nil, err - } - - attachments, err := s.attachWidgetUploads(ctx, &msg, req.AttachmentIDs) + attachments, err := s.createIncomingMessage(ctx, conversation, &msg, req.AttachmentIDs) if err != nil { return nil, err } - if _, err := EnqueueCaptainConversationResponseForMessage(ctx, s.worker, s.messageRepo.DB(), msg.ID); err != nil { - return nil, err - } - // Dispatch message.created event to trigger auto-reply listener + // Dispatch the same complete lifecycle context Chatwoot gives ActionCable. if s.dispatcher != nil { inbox, _ := s.inboxRepo.FindByID(ctx, conversation.InboxID) + data := map[string]interface{}{ + "inbox": inbox, + "conversation": conversation, + "contact": &contactInbox.Contact, + "widget_token": contactInbox.PubsubToken, + "channel_type": "web_widget", + } + if conversationCreated { + event := &channel.ChannelEvent{ + Type: channel.EventConversationCreated, Channel: channel.ChannelWebWidget, + ConversationID: conversation.ID, InboxID: conversation.InboxID, + AccountID: conversation.AccountID, ContactID: conversation.ContactID, + Timestamp: time.Now().Unix(), Data: data, + } + if dispatchErr := s.dispatcher.Dispatch(ctx, event); dispatchErr != nil { + applogger.L().Warnf("widget conversation event dispatch failed: inbox=%d conv=%d err=%v", + conversation.InboxID, conversation.ID, dispatchErr) + } + } + messageData := make(map[string]interface{}, len(data)+1) + for key, value := range data { + messageData[key] = value + } + messageData["message"] = &msg event := &channel.ChannelEvent{ Type: channel.EventMessageCreated, Channel: channel.ChannelWebWidget, @@ -433,13 +449,8 @@ func (s *WidgetService) SendMessage(ctx context.Context, req WidgetSendMessageRe InboxID: conversation.InboxID, AccountID: conversation.AccountID, Timestamp: time.Now().Unix(), - Data: map[string]interface{}{ - "inbox": inbox, - "content": msg.Content, - "sender_type": "Contact", - "channel_type": "web_widget", - "source_id": msg.SourceID, - }, + ContactID: conversation.ContactID, + Data: messageData, } if dispatchErr := s.dispatcher.Dispatch(ctx, event); dispatchErr != nil { applogger.L().Warnf("widget message event dispatch failed: inbox=%d conv=%d err=%v", @@ -457,18 +468,51 @@ func (s *WidgetService) SendMessage(ctx context.Context, req WidgetSendMessageRe }, nil } -func (s *WidgetService) reopenWidgetConversationForIncomingMessage(ctx context.Context, conversation *model.Conversation) error { - if conversation == nil || conversation.Muted { - return nil +func (s *WidgetService) createIncomingMessage(ctx context.Context, conversation *model.Conversation, message *model.Message, attachmentIDs []string) ([]model.Attachment, error) { + var messageTimestamp int64 + var attachments []model.Attachment + var captainJob *model.BackgroundJob + var captainJobCreated bool + reopen := conversation != nil && !conversation.Muted && + (conversation.Status == string(model.ConversationStatusSnoozed) || conversation.Status == string(model.ConversationStatusResolved)) + if err := s.messageRepo.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Create(message).Error; err != nil { + return err + } + var err error + attachments, err = s.attachWidgetUploadsInTransaction(ctx, tx, message, attachmentIDs) + if err != nil { + return err + } + messageTimestamp = message.CreatedAt.Unix() + updates := map[string]any{ + "last_activity_at": messageTimestamp, + "last_message_at": messageTimestamp, + } + if reopen { + updates["status"] = string(model.ConversationStatusOpen) + updates["snoozed_until"] = nil + } + if err := tx.Model(&model.Conversation{}). + Where("id = ? AND account_id = ?", conversation.ID, conversation.AccountID). + Updates(updates).Error; err != nil { + return err + } + captainJob, captainJobCreated, err = enqueueCaptainConversationResponseForMessageInTransaction(ctx, s.worker, tx, message.ID) + return err + }); err != nil { + return nil, fmt.Errorf("failed to create message: %w", err) } - if conversation.Status != string(model.ConversationStatusSnoozed) && conversation.Status != string(model.ConversationStatusResolved) { - return nil + conversation.LastActivityAt = &messageTimestamp + conversation.LastMessageAt = &messageTimestamp + if reopen { + conversation.Status = string(model.ConversationStatusOpen) + conversation.SnoozedUntil = nil } - conversation.Status = string(model.ConversationStatusOpen) - conversation.SnoozedUntil = nil - return s.conversationRepo.DB().WithContext(ctx).Model(&model.Conversation{}). - Where("id = ?", conversation.ID). - Updates(map[string]any{"status": conversation.Status, "snoozed_until": nil}).Error + if captainJobCreated { + s.worker.Publish(ctx, captainJob) + } + return attachments, nil } // GetConversations returns conversations for a widget contact. @@ -551,10 +595,13 @@ func (s *WidgetService) GetInboxMembersByWebsiteToken(ctx context.Context, websi if name == "" { name = member.User.Email } - availability := member.AvailabilityStatus + availability := member.AccountAvailability if availability == "" { availability = "offline" } + if !member.User.Active { + availability = "offline" + } payload = append(payload, WidgetInboxMember{ ID: member.User.ID, Name: name, @@ -993,6 +1040,10 @@ func (s *WidgetService) PublicCreateMessage(ctx context.Context, inboxIdentifier if len([]rune(req.Content)) > widgetMessageContentLimit { return nil, nil, nil, ErrWidgetMessageContentTooLong } + _, contactInbox, err := s.resolvePublicContactInbox(ctx, inboxIdentifier, sourceID) + if err != nil { + return nil, nil, nil, err + } conversation, err := s.PublicGetConversation(ctx, inboxIdentifier, sourceID, displayID) if err != nil { return nil, nil, nil, err @@ -1009,33 +1060,29 @@ func (s *WidgetService) PublicCreateMessage(ctx context.Context, inboxIdentifier Status: "sent", SourceID: req.EchoID, } - if err := s.messageRepo.Create(ctx, message); err != nil { - return nil, nil, nil, err - } - attachments, err := s.attachWidgetUploads(ctx, message, req.AttachmentIDs) + attachments, err := s.createIncomingMessage(ctx, conversation, message, req.AttachmentIDs) if err != nil { return nil, nil, nil, err } - if _, err := EnqueueCaptainConversationResponseForMessage(ctx, s.worker, s.messageRepo.DB(), message.ID); err != nil { - return nil, nil, nil, err - } - // Dispatch message.created event to trigger auto-reply listener + // Public API messages use the same committed lifecycle payload as widget messages. if s.dispatcher != nil { inbox, _ := s.inboxRepo.FindByID(ctx, conversation.InboxID) event := &channel.ChannelEvent{ Type: channel.EventMessageCreated, - Channel: channel.ChannelWebWidget, + Channel: channel.ChannelAPI, ConversationID: conversation.ID, InboxID: conversation.InboxID, AccountID: conversation.AccountID, + ContactID: conversation.ContactID, Timestamp: time.Now().Unix(), Data: map[string]interface{}{ "inbox": inbox, - "content": message.Content, - "sender_type": "Contact", - "channel_type": "web_widget", - "source_id": message.SourceID, + "conversation": conversation, + "contact": &contactInbox.Contact, + "widget_token": contactInbox.PubsubToken, + "channel_type": "api", + "message": message, }, } if dispatchErr := s.dispatcher.Dispatch(ctx, event); dispatchErr != nil { @@ -2039,6 +2086,16 @@ func (s *WidgetService) findPublicContact(ctx context.Context, accountID uint, r } func (s *WidgetService) attachWidgetUploads(ctx context.Context, message *model.Message, signedIDs []string) ([]model.Attachment, error) { + var attachments []model.Attachment + err := s.messageRepo.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var err error + attachments, err = s.attachWidgetUploadsInTransaction(ctx, tx, message, signedIDs) + return err + }) + return attachments, err +} + +func (s *WidgetService) attachWidgetUploadsInTransaction(ctx context.Context, tx *gorm.DB, message *model.Message, signedIDs []string) ([]model.Attachment, error) { if len(signedIDs) == 0 { return nil, nil } @@ -2049,7 +2106,7 @@ func (s *WidgetService) attachWidgetUploads(ctx context.Context, message *model. continue } var upload model.DirectUpload - if err := s.messageRepo.DB().WithContext(ctx).Where("upload_uuid = ?", signedID).First(&upload).Error; err != nil { + if err := tx.WithContext(ctx).Where("upload_uuid = ?", signedID).First(&upload).Error; err != nil { return nil, fmt.Errorf("direct upload not found: %w", err) } if upload.Source != model.DirectUploadSourceWidget { @@ -2063,7 +2120,7 @@ func (s *WidgetService) attachWidgetUploads(ctx context.Context, message *model. } if time.Now().After(upload.ExpiresAt) { upload.Status = model.DirectUploadStatusExpired - _ = s.messageRepo.DB().WithContext(ctx).Save(&upload).Error + _ = tx.WithContext(ctx).Save(&upload).Error return nil, errors.New("direct upload has expired") } attachment := model.Attachment{ @@ -2076,12 +2133,12 @@ func (s *WidgetService) attachWidgetUploads(ctx context.Context, message *model. FileName: upload.OriginalName, Metadata: string(upload.Metadata), } - if err := s.messageRepo.DB().WithContext(ctx).Create(&attachment).Error; err != nil { + if err := tx.WithContext(ctx).Create(&attachment).Error; err != nil { return nil, fmt.Errorf("failed to create attachment: %w", err) } upload.Status = model.DirectUploadStatusCompleted upload.AccountID = message.AccountID - if err := s.messageRepo.DB().WithContext(ctx).Save(&upload).Error; err != nil { + if err := tx.WithContext(ctx).Save(&upload).Error; err != nil { return nil, fmt.Errorf("failed to mark direct upload completed: %w", err) } attachments = append(attachments, attachment) diff --git a/backend/internal/service/widget_service_test.go b/backend/internal/service/widget_service_test.go index f2283a6c..bc2cc57e 100644 --- a/backend/internal/service/widget_service_test.go +++ b/backend/internal/service/widget_service_test.go @@ -16,11 +16,24 @@ import ( "gorm.io/gorm" "gorm.io/gorm/logger" + "github.com/gochat/gochat/internal/channel" "github.com/gochat/gochat/internal/model" + channelmodel "github.com/gochat/gochat/internal/model/channel" "github.com/gochat/gochat/internal/repository" "github.com/gochat/gochat/internal/worker" ) +type widgetLifecycleListener struct { + events []*channel.ChannelEvent +} + +func (l *widgetLifecycleListener) Name() string { return "widget-lifecycle-test" } + +func (l *widgetLifecycleListener) OnEvent(_ context.Context, event *channel.ChannelEvent) error { + l.events = append(l.events, event) + return nil +} + func TestIsGenericShangwutongName(t *testing.T) { assert.True(t, isGenericShangwutongName("商务通访客")) assert.True(t, isGenericShangwutongName("商务通访客·贵州贵阳")) @@ -48,6 +61,8 @@ func setupWidgetServiceTest(t *testing.T) (*gorm.DB, *WidgetService) { &model.ContactInbox{}, &model.Conversation{}, &model.Message{}, + &model.Attachment{}, + &model.DirectUpload{}, &model.WidgetThemeConfig{}, &model.PreChatForm{}, &model.WidgetFileUpload{}, @@ -58,6 +73,7 @@ func setupWidgetServiceTest(t *testing.T) (*gorm.DB, *WidgetService) { &model.CaptainPreference{}, &model.AgentBot{}, &model.AgentBotInbox{}, + &channelmodel.ChannelAPI{}, ), "failed to auto-migrate models") t.Cleanup(func() { @@ -94,6 +110,271 @@ func setupWidgetServiceTest(t *testing.T) (*gorm.DB, *WidgetService) { return db, svc } +func TestWidgetService_SendMessageDispatchesCompleteLifecycle(t *testing.T) { + db, svc := setupWidgetServiceTest(t) + seedWidgetInbox(t, db) + initResp, err := svc.Init(context.Background(), WidgetInitRequest{WebsiteToken: "test_ws_token_123"}) + require.NoError(t, err) + + listener := &widgetLifecycleListener{} + dispatcher := channel.NewDispatcher() + dispatcher.Register(listener) + svc.SetDispatcher(dispatcher) + + resp, err := svc.SendMessage(context.Background(), WidgetSendMessageRequest{ + WidgetToken: initResp.WidgetToken, + Content: "hello from widget", + }) + require.NoError(t, err) + require.Len(t, listener.events, 2) + assert.Equal(t, channel.EventConversationCreated, listener.events[0].Type) + assert.Equal(t, channel.EventMessageCreated, listener.events[1].Type) + assert.Equal(t, resp.ConversationID, listener.events[1].ConversationID) + assert.Equal(t, initResp.WidgetToken, listener.events[1].Data["widget_token"]) + assert.IsType(t, &model.Conversation{}, listener.events[1].Data["conversation"]) + assert.IsType(t, &model.Contact{}, listener.events[1].Data["contact"]) + assert.IsType(t, &model.Message{}, listener.events[1].Data["message"]) +} + +func TestWidgetService_SendMessageUpdatesExistingConversationActivityBeforeDispatch(t *testing.T) { + db, svc := setupWidgetServiceTest(t) + seedWidgetInbox(t, db) + initResp, err := svc.Init(context.Background(), WidgetInitRequest{WebsiteToken: "test_ws_token_123"}) + require.NoError(t, err) + + oldTimestamp := time.Now().Add(-time.Hour).Unix() + conversation := &model.Conversation{ + AccountID: initResp.AccountID, InboxID: initResp.InboxID, ContactID: initResp.ContactID, + ContactInboxID: &initResp.ContactInboxID, Status: string(model.ConversationStatusOpen), + ChannelType: string(model.InboxChannelTypeWebWidget), Channel: string(model.InboxChannelTypeWebWidget), + LastActivityAt: &oldTimestamp, LastMessageAt: &oldTimestamp, + } + require.NoError(t, db.Create(conversation).Error) + + listener := &widgetLifecycleListener{} + dispatcher := channel.NewDispatcher() + dispatcher.Register(listener) + svc.SetDispatcher(dispatcher) + + resp, err := svc.SendMessage(context.Background(), WidgetSendMessageRequest{ + WidgetToken: initResp.WidgetToken, + Content: "new visitor message", + ConversationID: &conversation.ID, + }) + require.NoError(t, err) + require.Len(t, listener.events, 1) + + wantTimestamp := resp.Message.CreatedAt.Unix() + var persisted model.Conversation + require.NoError(t, db.First(&persisted, conversation.ID).Error) + require.NotNil(t, persisted.LastActivityAt) + require.NotNil(t, persisted.LastMessageAt) + assert.Equal(t, wantTimestamp, *persisted.LastActivityAt) + assert.Equal(t, wantTimestamp, *persisted.LastMessageAt) + + eventConversation, ok := listener.events[0].Data["conversation"].(*model.Conversation) + require.True(t, ok) + require.NotNil(t, eventConversation.LastActivityAt) + require.NotNil(t, eventConversation.LastMessageAt) + assert.Greater(t, *eventConversation.LastActivityAt, oldTimestamp) + assert.Equal(t, wantTimestamp, *eventConversation.LastActivityAt) + assert.Equal(t, wantTimestamp, *eventConversation.LastMessageAt) +} + +func TestWidgetService_PublicCreateMessageUpdatesConversationBeforeDispatch(t *testing.T) { + db, svc := setupWidgetServiceTest(t) + _, _, _, _, conversation, displayID, _ := seedPublicMessageTest(t, db, model.ConversationStatusOpen) + + listener := &widgetLifecycleListener{} + dispatcher := channel.NewDispatcher() + dispatcher.Register(listener) + svc.SetDispatcher(dispatcher) + + message, _, _, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, PublicMessageRequest{Content: "hello"}) + require.NoError(t, err) + require.Len(t, listener.events, 1) + + wantTimestamp := message.CreatedAt.Unix() + var persisted model.Conversation + require.NoError(t, db.First(&persisted, conversation.ID).Error) + require.NotNil(t, persisted.LastActivityAt) + require.NotNil(t, persisted.LastMessageAt) + assert.Equal(t, wantTimestamp, *persisted.LastActivityAt) + assert.Equal(t, wantTimestamp, *persisted.LastMessageAt) + + event := listener.events[0] + assert.Equal(t, channel.EventMessageCreated, event.Type) + assert.Equal(t, channel.ChannelAPI, event.Channel) + assert.Equal(t, "visitor-token", event.Data["widget_token"]) + assert.Same(t, message, event.Data["message"]) + assert.IsType(t, &model.Contact{}, event.Data["contact"]) + eventConversation, ok := event.Data["conversation"].(*model.Conversation) + require.True(t, ok) + require.NotNil(t, eventConversation.LastActivityAt) + assert.Equal(t, wantTimestamp, *eventConversation.LastActivityAt) +} + +func TestWidgetService_PublicCreateMessageRollsBackSecondAttachmentFailureAndRetries(t *testing.T) { + db, svc := setupWidgetServiceTest(t) + account, _, _, _, conversation, displayID, oldTimestamp := seedPublicMessageTest(t, db, model.ConversationStatusOpen) + firstUpload := seedWidgetDirectUpload(t, db, account.ID, "upload-one") + + listener := &widgetLifecycleListener{} + dispatcher := channel.NewDispatcher() + dispatcher.Register(listener) + svc.SetDispatcher(dispatcher) + req := PublicMessageRequest{Content: "hello", AttachmentIDs: []string{firstUpload.UploadUUID, "upload-two"}} + + _, _, _, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req) + require.Error(t, err) + assertPublicMessageRollback(t, db, conversation.ID, oldTimestamp, firstUpload.ID) + assert.Empty(t, listener.events) + + secondUpload := seedWidgetDirectUpload(t, db, account.ID, "upload-two") + message, _, attachments, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req) + require.NoError(t, err) + require.Len(t, attachments, 2) + assert.Len(t, listener.events, 1) + var messageCount int64 + require.NoError(t, db.Model(&model.Message{}).Where("conversation_id = ?", conversation.ID).Count(&messageCount).Error) + assert.Equal(t, int64(1), messageCount) + for _, uploadID := range []uint{firstUpload.ID, secondUpload.ID} { + var upload model.DirectUpload + require.NoError(t, db.First(&upload, uploadID).Error) + assert.Equal(t, model.DirectUploadStatusCompleted, upload.Status) + } + for _, attachment := range attachments { + assert.Equal(t, message.ID, attachment.MessageID) + } +} + +func TestWidgetService_PublicCreateMessageRollsBackReopenFailureAndRetries(t *testing.T) { + db, svc := setupWidgetServiceTest(t) + _, _, _, _, conversation, displayID, oldTimestamp := seedPublicMessageTest(t, db, model.ConversationStatusResolved) + require.NoError(t, db.Exec(`CREATE TRIGGER fail_public_reopen BEFORE UPDATE OF status ON conversations + WHEN NEW.id = `+fmt.Sprint(conversation.ID)+` AND NEW.status = 'open' + BEGIN SELECT RAISE(FAIL, 'reopen failed'); END`).Error) + + listener := &widgetLifecycleListener{} + dispatcher := channel.NewDispatcher() + dispatcher.Register(listener) + svc.SetDispatcher(dispatcher) + req := PublicMessageRequest{Content: "reopen me"} + + _, _, _, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req) + require.Error(t, err) + assertPublicMessageRollback(t, db, conversation.ID, oldTimestamp, 0) + assert.Empty(t, listener.events) + var rolledBack model.Conversation + require.NoError(t, db.First(&rolledBack, conversation.ID).Error) + assert.Equal(t, string(model.ConversationStatusResolved), rolledBack.Status) + + require.NoError(t, db.Exec("DROP TRIGGER fail_public_reopen").Error) + message, reopened, _, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req) + require.NoError(t, err) + assert.Equal(t, string(model.ConversationStatusOpen), reopened.Status) + assert.Equal(t, message.CreatedAt.Unix(), *reopened.LastActivityAt) + assert.Len(t, listener.events, 1) + var messageCount int64 + require.NoError(t, db.Model(&model.Message{}).Where("conversation_id = ?", conversation.ID).Count(&messageCount).Error) + assert.Equal(t, int64(1), messageCount) +} + +func TestWidgetService_PublicCreateMessageRollsBackCaptainJobFailureAndRetries(t *testing.T) { + db, svc := setupWidgetServiceTest(t) + account, inbox, _, _, conversation, displayID, oldTimestamp := seedPublicMessageTest(t, db, model.ConversationStatusPending) + assistant := &model.CaptainAssistant{AccountID: account.ID, Name: "Captain", Status: model.AssistantStatusActive, Config: json.RawMessage(`{}`)} + require.NoError(t, db.Create(assistant).Error) + require.NoError(t, db.Create(&model.CaptainInbox{AccountID: account.ID, InboxID: inbox.ID, AssistantID: assistant.ID}).Error) + require.NoError(t, db.Create(&model.CaptainPreference{AccountID: account.ID, AutoReplyEnabled: true}).Error) + svc.SetWorkerPool(worker.NewWorkerPool(db)) + upload := seedWidgetDirectUpload(t, db, account.ID, "captain-upload") + require.NoError(t, db.Exec(`CREATE TRIGGER fail_captain_job BEFORE INSERT ON background_jobs + WHEN NEW.job_type = '`+TaskTypeCaptainConversationResponseBuilder+`' + BEGIN SELECT RAISE(FAIL, 'captain enqueue failed'); END`).Error) + + listener := &widgetLifecycleListener{} + dispatcher := channel.NewDispatcher() + dispatcher.Register(listener) + svc.SetDispatcher(dispatcher) + req := PublicMessageRequest{Content: "ask captain", AttachmentIDs: []string{upload.UploadUUID}} + + _, _, _, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req) + require.Error(t, err) + assertPublicMessageRollback(t, db, conversation.ID, oldTimestamp, upload.ID) + assert.Empty(t, listener.events) + var rolledBack model.Conversation + require.NoError(t, db.First(&rolledBack, conversation.ID).Error) + assert.Equal(t, string(model.ConversationStatusPending), rolledBack.Status) + + require.NoError(t, db.Exec("DROP TRIGGER fail_captain_job").Error) + message, _, attachments, err := svc.PublicCreateMessage(context.Background(), "public-api", "visitor-source", displayID, req) + require.NoError(t, err) + require.Len(t, attachments, 1) + assert.Len(t, listener.events, 1) + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeCaptainConversationResponseBuilder).First(&job).Error) + assert.Contains(t, string(job.Payload), fmt.Sprintf(`"message_id":%d`, message.ID)) + var messageCount int64 + require.NoError(t, db.Model(&model.Message{}).Where("conversation_id = ?", conversation.ID).Count(&messageCount).Error) + assert.Equal(t, int64(1), messageCount) +} + +func seedPublicMessageTest(t *testing.T, db *gorm.DB, status model.ConversationStatus) (*model.Account, *model.Inbox, *model.Contact, *model.ContactInbox, *model.Conversation, uint, int64) { + t.Helper() + account := &model.Account{Name: "Public API Account", Locale: "en", Status: "active"} + require.NoError(t, db.Create(account).Error) + inbox := &model.Inbox{AccountID: account.ID, Name: "API Inbox", ChannelType: "api", ChannelID: 1, Enabled: true} + require.NoError(t, db.Create(inbox).Error) + require.NoError(t, db.Create(&channelmodel.ChannelAPI{InboxID: inbox.ID, Identifier: "public-api"}).Error) + contact := &model.Contact{AccountID: account.ID, Name: "Public Visitor"} + require.NoError(t, db.Create(contact).Error) + contactInbox := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, SourceID: "visitor-source", PubsubToken: "visitor-token"} + require.NoError(t, db.Create(contactInbox).Error) + displayID := uint(42) + oldTimestamp := time.Now().Add(-time.Hour).Unix() + conversation := &model.Conversation{ + AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, ContactInboxID: &contactInbox.ID, + DisplayID: &displayID, Status: string(status), ChannelType: "api", Channel: "api", + LastActivityAt: &oldTimestamp, LastMessageAt: &oldTimestamp, + } + require.NoError(t, db.Create(conversation).Error) + return account, inbox, contact, contactInbox, conversation, displayID, oldTimestamp +} + +func seedWidgetDirectUpload(t *testing.T, db *gorm.DB, accountID uint, uploadUUID string) *model.DirectUpload { + t.Helper() + upload := &model.DirectUpload{ + UploadUUID: uploadUUID, AccountID: accountID, Status: model.DirectUploadStatusPending, + Source: model.DirectUploadSourceWidget, OriginalName: uploadUUID + ".txt", FileType: "file", + FileSize: 10, FileURL: "/uploads/" + uploadUUID, ExpiresAt: time.Now().Add(time.Hour), + } + require.NoError(t, db.Create(upload).Error) + return upload +} + +func assertPublicMessageRollback(t *testing.T, db *gorm.DB, conversationID uint, oldTimestamp int64, uploadID uint) { + t.Helper() + var messageCount, attachmentCount, jobCount int64 + require.NoError(t, db.Model(&model.Message{}).Where("conversation_id = ?", conversationID).Count(&messageCount).Error) + require.NoError(t, db.Model(&model.Attachment{}).Count(&attachmentCount).Error) + require.NoError(t, db.Model(&model.BackgroundJob{}).Count(&jobCount).Error) + assert.Zero(t, messageCount) + assert.Zero(t, attachmentCount) + assert.Zero(t, jobCount) + var conversation model.Conversation + require.NoError(t, db.First(&conversation, conversationID).Error) + require.NotNil(t, conversation.LastActivityAt) + require.NotNil(t, conversation.LastMessageAt) + assert.Equal(t, oldTimestamp, *conversation.LastActivityAt) + assert.Equal(t, oldTimestamp, *conversation.LastMessageAt) + if uploadID != 0 { + var upload model.DirectUpload + require.NoError(t, db.First(&upload, uploadID).Error) + assert.Equal(t, model.DirectUploadStatusPending, upload.Status) + } +} + // seedWidgetInbox creates an Account + a web_widget Inbox with a known website_token // in its ChannelConfig JSON, suitable for Init/SendMessage tests. func seedWidgetInbox(t *testing.T, db *gorm.DB) (*model.Account, *model.Inbox) { diff --git a/backend/internal/wsevent/bridge_listener.go b/backend/internal/wsevent/bridge_listener.go index 40da9e51..e664ffad 100644 --- a/backend/internal/wsevent/bridge_listener.go +++ b/backend/internal/wsevent/bridge_listener.go @@ -66,7 +66,11 @@ func (l *BridgeListener) OnEvent(ctx context.Context, event *channel.ChannelEven payload["inbox_id"] = event.InboxID } - l.publisher.PublishEvent(event.AccountID, eventType, payload) + if pubsubToken, _ := event.Data["widget_token"].(string); pubsubToken != "" { + l.publisher.PublishWidgetEvent(event.AccountID, pubsubToken, eventType, payload) + } else { + l.publisher.PublishEvent(event.AccountID, eventType, payload) + } applogger.L().Debugf("ws_bridge: forwarded event %s for account %d", eventType, event.AccountID) return nil @@ -150,17 +154,72 @@ func messagePushPayload(message *model.Message, data map[string]interface{}) map conversationPayload["last_activity_at"] = *conversation.LastActivityAt } } + senderType := senderTypeName(message.SenderType) if contact, ok := eventContact(data); ok { - payload["sender"] = contactPushPayload(contact) conversationPayload["contact_inbox"] = map[string]interface{}{ "source_id": contact.Identifier, } + if senderType == "Contact" { + payload["sender"] = contactPushPayload(contact) + } } payload["conversation_id"] = conversationID payload["conversation"] = conversationPayload + if senderType == "User" || senderType == "AgentBot" { + if sender, ok := eventSender(data); ok { + expectedType := "user" + if senderType == "AgentBot" { + expectedType = "agent_bot" + } + if sender["type"] == expectedType { + payload["sender"] = sender + } + } + } + payload["sender_type"] = senderType return payload } +func eventSender(data map[string]interface{}) (map[string]interface{}, bool) { + switch sender := data["sender"].(type) { + case *model.User: + if sender == nil { + return nil, false + } + availableName := sender.DisplayName + if availableName == "" { + availableName = sender.Name + } + return map[string]interface{}{ + "id": sender.ID, "name": sender.Name, "available_name": availableName, + "avatar_url": sender.AvatarURL, "thumbnail": sender.AvatarURL, + "type": "user", + }, true + case model.User: + return eventSender(map[string]interface{}{"sender": &sender}) + case *model.AgentBot: + if sender == nil { + return nil, false + } + return sender.PushEventData(), true + case model.AgentBot: + return eventSender(map[string]interface{}{"sender": &sender}) + case *model.CaptainAssistant: + if sender == nil { + return nil, false + } + return map[string]interface{}{ + "id": sender.ID, "name": sender.Name, "type": "agent_bot", + }, true + case model.CaptainAssistant: + return eventSender(map[string]interface{}{"sender": &sender}) + case map[string]interface{}: + return sender, true + default: + return nil, false + } +} + func contactPushPayload(contact *model.Contact) map[string]interface{} { payload := modelMap(contact) payload["created_at"] = contact.CreatedAt.Unix() @@ -246,6 +305,20 @@ func messageTypeValue(value string) int { } } +func senderTypeName(value string) string { + trimmed := strings.TrimSpace(value) + switch strings.ToLower(trimmed) { + case "contact": + return "Contact" + case "agentbot", "agent_bot", "captain::assistant", "captainassistant", "captain_assistant": + return "AgentBot" + case "user": + return "User" + default: + return trimmed + } +} + func nonEmpty(value, fallback string) string { if strings.TrimSpace(value) == "" { return fallback diff --git a/backend/internal/wsevent/bridge_listener_test.go b/backend/internal/wsevent/bridge_listener_test.go index 9145db6e..890f7f69 100644 --- a/backend/internal/wsevent/bridge_listener_test.go +++ b/backend/internal/wsevent/bridge_listener_test.go @@ -14,6 +14,8 @@ import ( type captureHub struct { accountID uint data []byte + room string + roomData []byte } func (h *captureHub) SendToAccount(accountID uint, data []byte) { @@ -21,7 +23,89 @@ func (h *captureHub) SendToAccount(accountID uint, data []byte) { h.data = append([]byte(nil), data...) } -func (h *captureHub) SendToRoom(string, []byte) {} +func (h *captureHub) SendToRoom(room string, data []byte) { + h.room = room + h.roomData = append([]byte(nil), data...) +} + +func TestBridgeListenerRoutesWebWidgetEventsToDashboardAndVisitor(t *testing.T) { + event := channel.NewChannelEvent(channel.EventMessageCreated, channel.ChannelWebWidget, 1, 4) + event.Data["widget_token"] = "visitor-token" + event.Data["message"] = &model.Message{Base: model.Base{ID: 12}, AccountID: 1, InboxID: 4, ConversationID: 2, MessageType: "outgoing", SenderType: "User"} + event.Data["sender"] = &model.User{Base: model.Base{ID: 8}, Name: "Agent A"} + + hub := &captureHub{} + listener := New(wspkg.NewEventPublisherLocal(hub, nil)) + if err := listener.OnEvent(context.Background(), event); err != nil { + t.Fatalf("OnEvent failed: %v", err) + } + if hub.accountID != 1 || len(hub.data) == 0 { + t.Fatalf("dashboard account room did not receive event: account=%d", hub.accountID) + } + if hub.room != "pubsub_token_visitor-token" || len(hub.roomData) == 0 { + t.Fatalf("visitor room did not receive event: room=%q", hub.room) + } + var envelope map[string]interface{} + if err := json.Unmarshal(hub.roomData, &envelope); err != nil { + t.Fatalf("decode visitor event: %v", err) + } + payload := envelope["data"].(map[string]interface{}) + if payload["sender_type"] != "User" || payload["sender"].(map[string]interface{})["name"] != "Agent A" { + t.Fatalf("visitor reply sender contract mismatch: %#v", payload) + } +} + +func TestBridgeListenerPreservesWebWidgetSenderTypeContract(t *testing.T) { + tests := []struct { + senderType string + expectedType string + sender interface{} + expectedName string + }{ + {senderType: "AgentBot", expectedType: "AgentBot", sender: &model.AgentBot{ID: 7, Name: "Reply Bot"}, expectedName: "Reply Bot"}, + {senderType: "Captain::Assistant", expectedType: "AgentBot", sender: &model.CaptainAssistant{Base: model.Base{ID: 8}, Name: "Captain"}, expectedName: "Captain"}, + {senderType: "CaptainAssistant", expectedType: "AgentBot", sender: &model.CaptainAssistant{Base: model.Base{ID: 8}, Name: "Captain"}, expectedName: "Captain"}, + {senderType: "CustomSender", expectedType: "CustomSender"}, + } + for _, tt := range tests { + t.Run(tt.senderType, func(t *testing.T) { + event := channel.NewChannelEvent(channel.EventMessageCreated, channel.ChannelWebWidget, 1, 4) + event.Data["widget_token"] = "visitor-token" + event.Data["contact"] = &model.Contact{Base: model.Base{ID: 9}, Name: "Visitor", Identifier: "visitor-9"} + if tt.sender != nil { + event.Data["sender"] = tt.sender + } + event.Data["message"] = &model.Message{ + Base: model.Base{ID: 12}, AccountID: 1, InboxID: 4, ConversationID: 2, + MessageType: "outgoing", SenderType: tt.senderType, + } + + hub := &captureHub{} + listener := New(wspkg.NewEventPublisherLocal(hub, nil)) + if err := listener.OnEvent(context.Background(), event); err != nil { + t.Fatalf("OnEvent failed: %v", err) + } + var envelope map[string]interface{} + if err := json.Unmarshal(hub.roomData, &envelope); err != nil { + t.Fatalf("decode visitor event: %v", err) + } + payload := envelope["data"].(map[string]interface{}) + if payload["sender_type"] != tt.expectedType { + t.Fatalf("sender_type=%#v, want %q", payload["sender_type"], tt.expectedType) + } + if tt.expectedName == "" { + if _, exists := payload["sender"]; exists { + t.Fatalf("unknown sender must not reuse contact: %#v", payload["sender"]) + } + return + } + sender, ok := payload["sender"].(map[string]interface{}) + if !ok || sender["name"] != tt.expectedName || sender["type"] != "agent_bot" { + t.Fatalf("sender does not match sender_type: %#v", payload) + } + }) + } +} func TestBridgeListenerMessageCreatedUsesChatwootPushPayload(t *testing.T) { now := time.Unix(1_783_834_149, 0)