HH-581: restore WebChannel status and reply visibility (#145)

* fix(realtime): normalize web channel status events (HH-581) * fix(realtime): resolve legacy widget tokens safely (HH-581) --------- Co-authored-by: Rogee <rogee@ipao.vip>
2026-08-24 08:06:56 +08:00
parent 84a1e35294
commit 4204ae6a3a
6 changed files with 179 additions and 19 deletions
@@ -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.
@@ -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)
@@ -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)
@@ -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
@@ -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
 	}
@@ -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)