diff --git a/backend/internal/repository/contact_inbox_repo.go b/backend/internal/repository/contact_inbox_repo.go index 1913c207..2a91b697 100644 --- a/backend/internal/repository/contact_inbox_repo.go +++ b/backend/internal/repository/contact_inbox_repo.go @@ -39,14 +39,22 @@ func (r *ContactInboxRepo) FindByContactAndInbox(ctx context.Context, contactID, return &ci, nil } -// CountByContactAndInbox returns the number of channel identities for a contact -// in one inbox. Historical conversations without contact_inbox_id are safe to -// resolve only when this count is exactly one. -func (r *ContactInboxRepo) CountByContactAndInbox(ctx context.Context, contactID, inboxID uint) (int64, error) { - var count int64 - err := r.db.WithContext(ctx).Model(&model.ContactInbox{}). - Where("contact_id = ? AND inbox_id = ?", contactID, inboxID).Count(&count).Error - return count, err +// ResolveForConversation returns the explicit channel identity, or the sole +// identity for a legacy conversation without contact_inbox_id. Ambiguous +// legacy ownership fails closed. +func (r *ContactInboxRepo) ResolveForConversation(ctx context.Context, contactInboxID *uint, contactID, inboxID uint) (*model.ContactInbox, error) { + query := r.db.WithContext(ctx).Where("contact_id = ? AND inbox_id = ?", contactID, inboxID) + if contactInboxID != nil { + query = query.Where("id = ?", *contactInboxID) + } + var contactInboxes []model.ContactInbox + if err := query.Limit(2).Find(&contactInboxes).Error; err != nil { + return nil, err + } + if len(contactInboxes) != 1 { + return nil, gorm.ErrRecordNotFound + } + return &contactInboxes[0], nil } // FindByContactInboxSource retrieves a contact_inbox by the Chatwoot builder identity. diff --git a/backend/internal/service/conversation_service.go b/backend/internal/service/conversation_service.go index e6e30c80..de5cf535 100644 --- a/backend/internal/service/conversation_service.go +++ b/backend/internal/service/conversation_service.go @@ -111,6 +111,16 @@ func (s *ConversationService) dispatchConversationEventWithData(ctx context.Cont event.Data[key] = value } event.Data["conversation"] = conversation + if eventType == channel.EventConversationOpened || eventType == channel.EventConversationResolved { + contactInbox, err := repository.NewContactInboxRepo(s.repo.DB()).ResolveForConversation( + ctx, conversation.ContactInboxID, conversation.ContactID, conversation.InboxID, + ) + if err != nil { + applogger.L().Warnf("failed to resolve contact inbox for conversation %d status event: %v", conversation.ID, err) + } else if strings.TrimSpace(contactInbox.PubsubToken) != "" { + event.Data["widget_token"] = contactInbox.PubsubToken + } + } applogger.L().Infof("dispatching event %s for conversation %d", eventType, conversation.ID) if err := s.dispatcher.Dispatch(ctx, event); err != nil { applogger.L().Errorf("failed to dispatch event %s for conversation %d: %v", eventType, conversation.ID, err) diff --git a/backend/internal/service/conversation_service_test.go b/backend/internal/service/conversation_service_test.go index 696623e0..47b07ade 100644 --- a/backend/internal/service/conversation_service_test.go +++ b/backend/internal/service/conversation_service_test.go @@ -2,9 +2,13 @@ package service import ( "context" + "encoding/json" + "fmt" "testing" "time" + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "gorm.io/datatypes" @@ -17,6 +21,8 @@ import ( "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/repository" "github.com/gochat/gochat/internal/worker" + wspkg "github.com/gochat/gochat/internal/ws" + "github.com/gochat/gochat/internal/wsevent" ) // ========== Test Setup ========== @@ -1077,6 +1083,105 @@ func TestConversationService_MutationEventsCarryChatwootChangeData(t *testing.T) assert.Equal(t, team.ID, *assignedConversation.TeamID) } +func TestConversationService_ToggleStatusCarriesWebWidgetToken(t *testing.T) { + tests := []struct { + name string + initialStatus string + status string + linked bool + ambiguous bool + wantToken bool + }{ + {name: "linked opened", initialStatus: "resolved", status: "open", linked: true, wantToken: true}, + {name: "legacy opened", initialStatus: "resolved", status: "open", wantToken: true}, + {name: "legacy resolved", initialStatus: "open", status: "resolved", wantToken: true}, + {name: "ambiguous legacy fails closed", initialStatus: "resolved", status: "open", ambiguous: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + svc, db := setupConversationService(t) + listener := &captureConversationEventsListener{} + svc.dispatcher.Register(listener) + + account := createConversationServiceTestAccount(t, db) + inbox := createConversationServiceTestInbox(t, db, account.ID) + contact := createConversationServiceTestContact(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) + if tt.ambiguous { + require.NoError(t, db.Create(&model.ContactInbox{ + ContactID: contact.ID, InboxID: inbox.ID, SourceID: "other-visitor", PubsubToken: "other-token", + }).Error) + } + conversation := createConversationServiceTestConversation(t, db, account.ID, inbox.ID, contact.ID, tt.initialStatus) + if tt.linked { + require.NoError(t, db.Model(conversation).Update("contact_inbox_id", contactInbox.ID).Error) + } + + _, err := svc.ToggleStatus(context.Background(), account.ID, conversation.ID, ToggleStatusRequest{Status: tt.status}) + require.NoError(t, err) + require.NotEmpty(t, listener.events) + if tt.wantToken { + assert.Equal(t, "visitor-token", listener.events[len(listener.events)-1].Data["widget_token"]) + } else { + assert.NotContains(t, listener.events[len(listener.events)-1].Data, "widget_token") + } + }) + } +} + +func TestConversationService_LegacyStatusEventPublishesDurablyToRedis(t *testing.T) { + svc, db := setupConversationService(t) + mini := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mini.Addr()}) + t.Cleanup(func() { require.NoError(t, rdb.Close()) }) + pool := worker.NewWorkerPool(db) + publisher := wspkg.NewEventPublisher(nil, nil, wspkg.NewBroadcastRelay(rdb, nil)) + publisher.SetWorkerPool(pool) + svc.dispatcher.Register(wsevent.New(publisher)) + + account := createConversationServiceTestAccount(t, db) + inbox := createConversationServiceTestInbox(t, db, account.ID) + contact := createConversationServiceTestContact(t, db, account.ID) + require.NoError(t, db.Create(&model.ContactInbox{ + ContactID: contact.ID, InboxID: inbox.ID, SourceID: "legacy-visitor", PubsubToken: "legacy-token", + }).Error) + conversation := createConversationServiceTestConversation(t, db, account.ID, inbox.ID, contact.ID, "resolved") + + accountChannel := wspkg.RedisPrefixRoom + "account_" + fmt.Sprint(account.ID) + tokenChannel := wspkg.RedisPrefixRoom + "pubsub_token_legacy-token" + sub := rdb.Subscribe(context.Background(), accountChannel, tokenChannel) + t.Cleanup(func() { require.NoError(t, sub.Close()) }) + require.NoError(t, sub.Ping(context.Background())) + + _, err := svc.ToggleStatus(context.Background(), account.ID, conversation.ID, ToggleStatusRequest{Status: "open"}) + require.NoError(t, err) + var jobs int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("queue = ?", "events").Count(&jobs).Error) + require.Equal(t, int64(2), jobs) + for range 2 { + processed, processErr := pool.ProcessOne(context.Background()) + require.True(t, processed) + require.NoError(t, processErr) + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + received := map[string]bool{} + for range 2 { + message, receiveErr := sub.ReceiveMessage(ctx) + require.NoError(t, receiveErr) + var envelope map[string]interface{} + require.NoError(t, json.Unmarshal([]byte(message.Payload), &envelope)) + assert.Equal(t, wspkg.EventConversationStatusChanged, envelope["event"]) + assert.Equal(t, "open", envelope["data"].(map[string]interface{})["status"]) + received[message.Channel] = true + } + assert.Equal(t, map[string]bool{accountChannel: true, tokenChannel: true}, received) +} + func TestConversationService_ToggleTypingDispatchesChatwootPayload(t *testing.T) { db := setupConversationServiceTestDB(t) convRepo := repository.NewConversationRepo(db) diff --git a/backend/internal/service/widget_service.go b/backend/internal/service/widget_service.go index 8512668e..c0756631 100644 --- a/backend/internal/service/widget_service.go +++ b/backend/internal/service/widget_service.go @@ -578,12 +578,12 @@ func (s *WidgetService) GetConversation(ctx context.Context, widgetToken string, } func (s *WidgetService) findWidgetConversations(ctx context.Context, contactInbox *model.ContactInbox, offset, limit int) ([]model.Conversation, int64, error) { - count, err := s.contactInboxRepo.CountByContactAndInbox(ctx, contactInbox.ContactID, contactInbox.InboxID) - if err != nil { + legacyOwner, err := s.contactInboxRepo.ResolveForConversation(ctx, nil, contactInbox.ContactID, contactInbox.InboxID) + if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { return nil, 0, err } return s.conversationRepo.FindByContactInbox(ctx, contactInbox.Contact.AccountID, contactInbox.ContactID, - contactInbox.InboxID, contactInbox.ID, count == 1, offset, limit) + contactInbox.InboxID, contactInbox.ID, err == nil && legacyOwner.ID == contactInbox.ID, offset, limit) } func (s *WidgetService) resolveWidgetConversation(ctx context.Context, contactInbox *model.ContactInbox, conversationID uint) (*model.Conversation, error) { @@ -594,17 +594,14 @@ func (s *WidgetService) resolveWidgetConversation(ctx context.Context, contactIn if conversation.ContactID != contactInbox.ContactID || conversation.InboxID != contactInbox.InboxID { return nil, errWidgetConversationOwnership } - if conversation.ContactInboxID != nil { - if *conversation.ContactInboxID == contactInbox.ID { - return conversation, nil - } - return nil, errWidgetConversationOwnership - } - count, err := s.contactInboxRepo.CountByContactAndInbox(ctx, contactInbox.ContactID, contactInbox.InboxID) + owner, err := s.contactInboxRepo.ResolveForConversation(ctx, conversation.ContactInboxID, contactInbox.ContactID, contactInbox.InboxID) if err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, errWidgetConversationOwnership + } return nil, err } - if count != 1 { + if owner.ID != contactInbox.ID { return nil, errWidgetConversationOwnership } return conversation, nil diff --git a/backend/internal/wsevent/bridge_listener.go b/backend/internal/wsevent/bridge_listener.go index 42b35c8d..694f8c91 100644 --- a/backend/internal/wsevent/bridge_listener.go +++ b/backend/internal/wsevent/bridge_listener.go @@ -87,6 +87,9 @@ func realtimeEvent(event *channel.ChannelEvent) (uint, string, string, map[strin return 0, "", "", nil, false } eventType := string(event.Type) + if event.Type == channel.EventConversationOpened || event.Type == channel.EventConversationResolved { + eventType = wspkg.EventConversationStatusChanged + } if !isWSEventType(eventType) { return 0, "", "", nil, false } diff --git a/backend/internal/wsevent/bridge_listener_test.go b/backend/internal/wsevent/bridge_listener_test.go index a3183d86..14fa65f3 100644 --- a/backend/internal/wsevent/bridge_listener_test.go +++ b/backend/internal/wsevent/bridge_listener_test.go @@ -9,6 +9,8 @@ import ( "github.com/gochat/gochat/internal/channel" "github.com/gochat/gochat/internal/model" wspkg "github.com/gochat/gochat/internal/ws" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) type captureHub struct { @@ -60,6 +62,41 @@ func TestBridgeListenerRoutesWebWidgetEventsToDashboardAndVisitor(t *testing.T) } } +func TestBridgeListenerPublishesConversationStatusChangesToDashboardAndVisitor(t *testing.T) { + for _, tt := range []struct { + eventType channel.EventType + status string + }{ + {eventType: channel.EventConversationOpened, status: "open"}, + {eventType: channel.EventConversationResolved, status: "resolved"}, + } { + t.Run(tt.status, func(t *testing.T) { + displayID := uint(42) + event := channel.NewChannelEvent(tt.eventType, channel.ChannelWebWidget, 1, 4) + event.Data["widget_token"] = "visitor-token" + event.Data["conversation"] = &model.Conversation{ + Base: model.Base{ID: 2}, DisplayID: &displayID, Status: tt.status, + } + + hub := &captureHub{} + listener := New(wspkg.NewEventPublisherLocal(hub, nil)) + require.NoError(t, listener.OnEvent(context.Background(), event)) + + for audience, raw := range map[string][]byte{ + "dashboard": hub.data, + "visitor": hub.roomData, + } { + var envelope map[string]interface{} + require.NoError(t, json.Unmarshal(raw, &envelope), audience) + assert.Equal(t, wspkg.EventConversationStatusChanged, envelope["event"], audience) + payload := envelope["data"].(map[string]interface{}) + assert.Equal(t, tt.status, payload["status"], audience) + assert.Equal(t, float64(displayID), payload["id"], audience) + } + }) + } +} + func TestBridgeListenerReturnsPublisherError(t *testing.T) { event := channel.NewChannelEvent(channel.EventInboxCreated, channel.ChannelAPI, 1, 4) event.Data["invalid"] = make(chan int)