From b31b1b9562cfb60fc8717d9502ef35546ae5a188 Mon Sep 17 00:00:00 2001 From: Rogee Date: Thu, 20 Aug 2026 22:22:27 +0800 Subject: [PATCH] H-337: restore Web widget reply visibility (#64) * H-337: restore widget reply delivery * H-337: harden widget conversation ownership --------- Co-authored-by: Rogee --- .../widget_conversation_migration_test.go | 72 +++++ .../internal/handler/widget/widget_handler.go | 36 ++- .../handler/widget/widget_handler_test.go | 75 +++++- backend/internal/handler/ws/handler.go | 30 ++- backend/internal/handler/ws/hub.go | 19 +- backend/internal/handler/ws/hub_extra_test.go | 4 +- backend/internal/handler/ws/protocol.go | 5 +- backend/internal/handler/ws/ws_test.go | 250 ++++++++++++++++++ .../internal/repository/contact_inbox_repo.go | 10 + .../internal/repository/conversation_repo.go | 16 ++ .../repository/conversation_repo_test.go | 27 ++ backend/internal/service/widget_service.go | 73 ++--- .../internal/service/widget_service_test.go | 133 ++++++++++ backend/internal/ws/auth.go | 40 +-- backend/internal/ws/coverage4_test.go | 2 +- backend/internal/ws/coverage5_test.go | 2 +- backend/internal/ws/coverage7_test.go | 2 +- .../internal/wsevent/bridge_listener_test.go | 5 + ...dget_conversation_contact_inboxes.down.sql | 10 + ...widget_conversation_contact_inboxes.up.sql | 24 ++ .../helpers/BaseActionCableConnector.js | 2 + .../specs/BaseActionCableConnector.spec.js | 93 +++++++ 22 files changed, 847 insertions(+), 83 deletions(-) create mode 100644 backend/internal/database/widget_conversation_migration_test.go create mode 100644 backend/migrations/000082_backfill_widget_conversation_contact_inboxes.down.sql create mode 100644 backend/migrations/000082_backfill_widget_conversation_contact_inboxes.up.sql create mode 100644 frontend/app/javascript/shared/helpers/specs/BaseActionCableConnector.spec.js diff --git a/backend/internal/database/widget_conversation_migration_test.go b/backend/internal/database/widget_conversation_migration_test.go new file mode 100644 index 00000000..0c0ed08e --- /dev/null +++ b/backend/internal/database/widget_conversation_migration_test.go @@ -0,0 +1,72 @@ +package database + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +const widgetConversationMigrationBaseline = ` +CREATE TABLE inboxes (id INTEGER PRIMARY KEY, channel_type TEXT NOT NULL); +CREATE TABLE contact_inboxes (id INTEGER PRIMARY KEY, contact_id INTEGER NOT NULL, inbox_id INTEGER NOT NULL); +CREATE TABLE conversations (id INTEGER PRIMARY KEY, contact_id INTEGER NOT NULL, inbox_id INTEGER NOT NULL, contact_inbox_id INTEGER); +` + +func TestWidgetConversationContactInboxMigrationRoundTrip(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "widget-conversations.db") + dbURL := "sqlite3://" + dbPath + db, err := gorm.Open(sqlite.Open(dbPath), &gorm.Config{}) + require.NoError(t, err) + sqlDB, err := db.DB() + require.NoError(t, err) + sqlDB.SetMaxOpenConns(1) + t.Cleanup(func() { _ = sqlDB.Close() }) + require.NoError(t, db.Exec(widgetConversationMigrationBaseline).Error) + require.NoError(t, db.Exec(` +INSERT INTO inboxes VALUES (1, 'web_widget'), (2, 'api'); +INSERT INTO contact_inboxes VALUES (1, 10, 1), (2, 20, 1), (3, 20, 1), (4, 30, 1); +INSERT INTO conversations VALUES + (1, 10, 1, NULL), + (2, 20, 1, NULL), + (3, 30, 1, 4), + (4, 10, 2, NULL); +`).Error) + + migrations := t.TempDir() + for _, direction := range []string{"up", "down"} { + require.NoError(t, os.WriteFile(filepath.Join(migrations, "000081_baseline."+direction+".sql"), []byte("SELECT 1;"), 0o600)) + name := "000082_backfill_widget_conversation_contact_inboxes." + direction + ".sql" + data, readErr := os.ReadFile(filepath.Join("..", "..", "migrations", name)) + require.NoError(t, readErr) + require.NoError(t, os.WriteFile(filepath.Join(migrations, name), data, 0o600)) + } + require.NoError(t, ForceVersion(dbURL, migrations, 81)) + require.NoError(t, MigrateSteps(dbURL, migrations, 1)) + + var bindings []struct { + ID uint + ContactInboxID *uint + } + require.NoError(t, db.Raw("SELECT id, contact_inbox_id FROM conversations ORDER BY id").Scan(&bindings).Error) + require.Len(t, bindings, 4) + require.NotNil(t, bindings[0].ContactInboxID) + assert.Equal(t, uint(1), *bindings[0].ContactInboxID) + assert.Nil(t, bindings[1].ContactInboxID) + require.NotNil(t, bindings[2].ContactInboxID) + assert.Equal(t, uint(4), *bindings[2].ContactInboxID) + assert.Nil(t, bindings[3].ContactInboxID) + + require.NoError(t, MigrateSteps(dbURL, migrations, -1)) + require.NoError(t, db.Raw("SELECT id, contact_inbox_id FROM conversations ORDER BY id").Scan(&bindings).Error) + assert.Nil(t, bindings[0].ContactInboxID) + assert.Nil(t, bindings[1].ContactInboxID) + require.NotNil(t, bindings[2].ContactInboxID) + assert.Equal(t, uint(4), *bindings[2].ContactInboxID) + assert.Nil(t, bindings[3].ContactInboxID) + assert.False(t, db.Migrator().HasTable("widget_conversation_contact_inbox_backfills")) +} diff --git a/backend/internal/handler/widget/widget_handler.go b/backend/internal/handler/widget/widget_handler.go index 6a56c30b..cb89f52c 100644 --- a/backend/internal/handler/widget/widget_handler.go +++ b/backend/internal/handler/widget/widget_handler.go @@ -204,7 +204,11 @@ func (h *WidgetHandler) SendMessage(c *gin.Context) { c.JSON(http.StatusOK, resp) return } - payload := widgetMessagePayload(resp.Message, resp.ConversationID) + conversationID := resp.ConversationID + if conversation, lookupErr := h.widgetService.GetConversation(c.Request.Context(), widgetToken, resp.ConversationID); lookupErr == nil { + conversationID = widgetConversationID(*conversation) + } + payload := widgetMessagePayload(resp.Message, conversationID) if len(resp.Attachments) > 0 { payload["attachments"] = widgetAttachmentPayloads(resp.Attachments) } @@ -271,8 +275,12 @@ func (h *WidgetHandler) GetLatestMessages(c *gin.Context) { } payload := make([]gin.H, 0, len(messages)) + conversationID := uint(0) + if conversation != nil { + conversationID = widgetConversationID(*conversation) + } for _, msg := range messages { - messagePayload := widgetMessagePayload(msg, msg.ConversationID) + messagePayload := widgetMessagePayload(msg, conversationID) if attachments, err := h.widgetService.GetMessageAttachments(c.Request.Context(), msg.ID); err == nil && len(attachments) > 0 { messagePayload["attachments"] = widgetAttachmentPayloads(attachments) } @@ -350,7 +358,7 @@ func (h *WidgetHandler) CreateConversation(c *gin.Context) { return } payload := widgetConversationPayload(*conversation) - payload["messages"] = []gin.H{widgetMessagePayload(resp.Message, resp.ConversationID)} + payload["messages"] = []gin.H{widgetMessagePayload(resp.Message, widgetConversationID(*conversation))} c.JSON(http.StatusOK, payload) } @@ -1275,7 +1283,7 @@ func widgetMessagePayload(message model.Message, conversationID uint) gin.H { "content": message.Content, "inbox_id": message.InboxID, "conversation_id": conversationID, - "message_type": message.MessageType, + "message_type": widgetMessageType(message.MessageType), "content_type": message.ContentType, "content_attributes": webhookutil.SanitizeOutboundJSON(message.ContentAttributes), "created_at": message.CreatedAt.Unix(), @@ -1284,6 +1292,26 @@ func widgetMessagePayload(message model.Message, conversationID uint) gin.H { } } +func widgetMessageType(value string) int { + switch strings.ToLower(strings.TrimSpace(value)) { + case "incoming": + return 0 + case "activity": + return 2 + case "template": + return 3 + default: + return 1 + } +} + +func widgetConversationID(conversation model.Conversation) uint { + if conversation.DisplayID != nil && *conversation.DisplayID != 0 { + return *conversation.DisplayID + } + return conversation.ID +} + func widgetAttachmentPayloads(attachments []model.Attachment) []gin.H { payload := make([]gin.H, 0, len(attachments)) for _, attachment := range attachments { diff --git a/backend/internal/handler/widget/widget_handler_test.go b/backend/internal/handler/widget/widget_handler_test.go index d6192350..5eef7814 100644 --- a/backend/internal/handler/widget/widget_handler_test.go +++ b/backend/internal/handler/widget/widget_handler_test.go @@ -482,7 +482,7 @@ func TestWidgetHandler_ChatwootConversationQueryTokenReusesSession(t *testing.T) var messageResp map[string]interface{} require.NoError(t, json.Unmarshal(wMessage.Body.Bytes(), &messageResp)) assert.Equal(t, "Popout session message", messageResp["content"]) - assert.Equal(t, "incoming", messageResp["message_type"]) + assert.Equal(t, float64(0), messageResp["message_type"]) assertWidgetMessageFixtureShape(t, messageResp) wLatest := httptest.NewRecorder() @@ -740,6 +740,22 @@ func TestWidgetHandler_ChatwootMessages_AuthTokenAndNestedPayload(t *testing.T) assert.Nil(t, messageResp["message"]) assert.Equal(t, "Hello from Chatwoot widget", messageResp["content"]) assert.NotEmpty(t, messageResp["conversation_id"]) + assert.Equal(t, float64(0), messageResp["message_type"]) + + var conversation model.Conversation + require.NoError(t, db.First(&conversation).Error) + displayID := uint(42) + require.NoError(t, db.Model(&conversation).Update("display_id", displayID).Error) + require.NoError(t, db.Create(&model.Message{ + ConversationID: conversation.ID, + AccountID: conversation.AccountID, + InboxID: conversation.InboxID, + Content: "Dashboard reply visible after refresh", + ContentType: "text", + MessageType: string(model.MessageTypeOutgoing), + SenderType: string(model.SenderTypeUser), + Status: "sent", + }).Error) wIndex := httptest.NewRecorder() reqIndex, _ := http.NewRequest("GET", "/api/v1/widget/messages", nil) @@ -750,10 +766,14 @@ func TestWidgetHandler_ChatwootMessages_AuthTokenAndNestedPayload(t *testing.T) var indexResp map[string]interface{} require.NoError(t, json.Unmarshal(wIndex.Body.Bytes(), &indexResp)) payload := indexResp["payload"].([]interface{}) - require.Len(t, payload, 1) + require.Len(t, payload, 2) firstMessage := payload[0].(map[string]interface{}) assertWidgetMessageFixtureShape(t, firstMessage) assert.Equal(t, "Hello from Chatwoot widget", firstMessage["content"]) + reply := payload[1].(map[string]interface{}) + assert.Equal(t, "Dashboard reply visible after refresh", reply["content"]) + assert.Equal(t, float64(1), reply["message_type"]) + assert.Equal(t, float64(displayID), reply["conversation_id"]) wContact := httptest.NewRecorder() reqContact, _ := http.NewRequest("GET", "/api/v1/widget/contact", nil) @@ -791,6 +811,57 @@ func TestWidgetHandler_ChatwootMessagesIndexFiltersInternalMessages(t *testing.T assert.Equal(t, float64(1), indexResp["meta"].(map[string]interface{})["total"]) } +func TestWidgetHandler_ExplicitConversationEntrypointsRejectSiblingToken(t *testing.T) { + db, router, _ := setupWidgetHandlerTest(t) + account, inbox := seedWidgetHandlerData(t, db) + contact := &model.Contact{AccountID: account.ID, Name: "Shared visitor"} + require.NoError(t, db.Create(contact).Error) + owner := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, PubsubToken: "owner-token"} + sibling := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, PubsubToken: "sibling-token"} + require.NoError(t, db.Create(owner).Error) + require.NoError(t, db.Create(sibling).Error) + conversation := &model.Conversation{ + AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, ContactInboxID: &owner.ID, + Status: "open", ChannelType: "web_widget", Channel: "web_widget", + } + require.NoError(t, db.Create(conversation).Error) + message := &model.Message{ + AccountID: account.ID, InboxID: inbox.ID, ConversationID: conversation.ID, + Content: "owner only", ContentType: "integrations", MessageType: "outgoing", Status: "sent", + ContentAttributes: datatypes.JSON(`{"data":{"meeting_id":"meeting-1"}}`), + } + require.NoError(t, db.Create(message).Error) + + tests := []struct { + name, method, path, body string + }{ + {"messages", http.MethodGet, fmt.Sprintf("/widget/conversations/%d/messages", conversation.ID), ""}, + {"typing", http.MethodPost, fmt.Sprintf("/widget/conversations/%d/toggle_typing", conversation.ID), `{"typing":true}`}, + {"message update", http.MethodPatch, fmt.Sprintf("/api/v1/widget/messages/%d", message.ID), `{"message":{"submitted_values":[{"value":"nope"}]}}`}, + {"dyte", http.MethodPost, "/api/v1/widget/integrations/dyte/add_participant_to_meeting?website_token=handler_ws_token_123", fmt.Sprintf(`{"message_id":%d}`, message.ID)}, + {"send", http.MethodPost, "/widget/messages", fmt.Sprintf(`{"content":"nope","conversation_id":%d}`, conversation.ID)}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := httptest.NewRequest(tt.method, tt.path, strings.NewReader(tt.body)) + req.Header.Set("X-Widget-Token", sibling.PubsubToken) + if tt.body != "" { + req.Header.Set("Content-Type", "application/json") + } + response := httptest.NewRecorder() + router.ServeHTTP(response, req) + assert.GreaterOrEqual(t, response.Code, http.StatusBadRequest, response.Body.String()) + }) + } + + var messageCount int64 + require.NoError(t, db.Model(&model.Message{}).Where("conversation_id = ?", conversation.ID).Count(&messageCount).Error) + assert.Equal(t, int64(1), messageCount) + var persisted model.Message + require.NoError(t, db.First(&persisted, message.ID).Error) + assert.JSONEq(t, `{"data":{"meeting_id":"meeting-1"}}`, string(persisted.ContentAttributes)) +} + func TestWidgetHandler_ChatwootMessagesIndexReturnsLatestTwenty(t *testing.T) { db, router, _ := setupWidgetHandlerTest(t) _, inbox := seedWidgetHandlerData(t, db) diff --git a/backend/internal/handler/ws/handler.go b/backend/internal/handler/ws/handler.go index b4b5fb6d..5b548f4e 100644 --- a/backend/internal/handler/ws/handler.go +++ b/backend/internal/handler/ws/handler.go @@ -26,7 +26,7 @@ func uintToStr(u uint) string { // // Supports two authentication paths: // 1. Agent/User auth: JWT token (from query param or Authorization header) -// 2. Contact auth: pubsub_token + user_id (Chatwoot RoomChannel pattern) +// 2. Contact auth: pubsub_token (Chatwoot RoomChannel pattern) type Handler struct { hub *Hub authenticator *wspkg.WSAuthenticator @@ -55,7 +55,7 @@ func NewHandler(hub *Hub, authenticator *wspkg.WSAuthenticator) *Handler { } // ServeWS handles the WebSocket upgrade request at /ws. -// URL: /ws?token= OR /ws?pubsub_token=&user_id= +// URL: /ws?token= OR /ws?pubsub_token= // On successful upgrade, the handler: // 1. Authenticates the request via WSAuthenticator (JWT or pubsub_token) // 2. Authorizes the user/contact for the requested account @@ -272,8 +272,19 @@ func (h *Handler) handleSubscribe(client *Client, cmd CommandFrame) { return } - // Validate account_id matches the client's authenticated account - if identifier.AccountID != client.AccountID { + room := "" + if client.IsContact { + if identifier.Channel != ChannelRoom || identifier.PubsubToken == "" || identifier.PubsubToken != client.PubsubToken { + rejectData, _ := json.Marshal(RejectFrame{ + Type: ServerRejectSubscribe, + Identifier: cmd.Identifier, + Reason: "invalid contact RoomChannel subscription", + }) + client.Send <- rejectData + return + } + room = pubsubTokenRoomName(identifier.PubsubToken) + } else if identifier.AccountID != client.AccountID { rejectData, _ := json.Marshal(RejectFrame{ Type: ServerRejectSubscribe, Identifier: cmd.Identifier, @@ -283,14 +294,15 @@ func (h *Handler) handleSubscribe(client *Client, cmd CommandFrame) { return } - // Determine room name based on channel type (uses Hub's canonical naming) - room := "" - switch identifier.Channel { - case ChannelAccount, ChannelRoom: + // Determine room name based on channel type (uses Hub's canonical naming). + // Contact RoomChannel was resolved above from its authenticated token. + switch { + case room != "": + case identifier.Channel == ChannelAccount || identifier.Channel == ChannelRoom: // RoomChannel is Chatwoot's single-subscription model — it maps // to the account room (all account-level events are delivered). room = accountRoomName(identifier.AccountID) - case ChannelConversation: + case identifier.Channel == ChannelConversation: if identifier.ConversationID == 0 { rejectData, _ := json.Marshal(RejectFrame{ Type: ServerRejectSubscribe, diff --git a/backend/internal/handler/ws/hub.go b/backend/internal/handler/ws/hub.go index 26035f71..d315cd03 100644 --- a/backend/internal/handler/ws/hub.go +++ b/backend/internal/handler/ws/hub.go @@ -180,21 +180,20 @@ func (h *Hub) Register(c *Client) { h.clients[c.ID] = c - // Auto-subscribe to account room on connect - roomName := accountRoomName(c.AccountID) - h.subscribeClient(c.ID, roomName) - c.SubscribedRooms[roomName] = true + if c.IsContact && c.PubsubToken != "" { + tokenRoom := pubsubTokenRoomName(c.PubsubToken) + h.subscribeClient(c.ID, tokenRoom) + c.SubscribedRooms[tokenRoom] = true + } else if !c.IsContact { + roomName := accountRoomName(c.AccountID) + h.subscribeClient(c.ID, roomName) + c.SubscribedRooms[roomName] = true + } // Set up presence tracking if h.presenceMgr != nil { if c.IsContact { h.presenceMgr.OnContactConnect(context.Background(), c.ContactID, c.AccountID) - // Also auto-subscribe to pubsub_token room (Chatwoot RoomChannel pattern) - if c.PubsubToken != "" { - tokenRoom := pubsubTokenRoomName(c.PubsubToken) - h.subscribeClient(c.ID, tokenRoom) - c.SubscribedRooms[tokenRoom] = true - } } else { cancelFn := h.presenceMgr.OnAgentConnect(context.Background(), c.UserID, c.AccountID) c.CancelPresence = cancelFn diff --git a/backend/internal/handler/ws/hub_extra_test.go b/backend/internal/handler/ws/hub_extra_test.go index eefb1993..ad02cfcb 100644 --- a/backend/internal/handler/ws/hub_extra_test.go +++ b/backend/internal/handler/ws/hub_extra_test.go @@ -61,8 +61,8 @@ func TestHub_Register_ContactWithPresence(t *testing.T) { hub.Register(client) assert.Contains(t, hub.clients, client.ID) - // Contact should be auto-subscribed to account room + pubsub_token room - assert.True(t, client.SubscribedRooms[accountRoomName(10)]) + // Contacts only receive events addressed to their own pubsub token. + assert.False(t, client.SubscribedRooms[accountRoomName(10)]) assert.True(t, client.SubscribedRooms[pubsubTokenRoomName("token123")]) hub.Unregister(client) diff --git a/backend/internal/handler/ws/protocol.go b/backend/internal/handler/ws/protocol.go index 11fb3e11..0f15c9f1 100644 --- a/backend/internal/handler/ws/protocol.go +++ b/backend/internal/handler/ws/protocol.go @@ -34,7 +34,7 @@ const ( // ServerConfirmSubscribe acknowledges a successful subscription ServerConfirmSubscribe ServerMessageType = "confirm_subscription" // ServerConfirmUnsubscribe acknowledges a successful unsubscribe - ServerConfirmUnsubscribe ServerMessageType = "confirm_unsubscribe" // NOTE: ActionCable uses confirm_subscription for both sub and unsub + ServerConfirmUnsubscribe ServerMessageType = "confirm_unsubscribe" // NOTE: ActionCable uses confirm_subscription for both sub and unsub // ServerRejectSubscribe rejects a subscription attempt ServerRejectSubscribe ServerMessageType = "reject_subscription" // ServerPing is a heartbeat response @@ -61,13 +61,14 @@ type ChannelIdentifier struct { Channel string `json:"channel"` // "AccountChannel" or "ConversationChannel" AccountID uint `json:"account_id"` // required for both channels ConversationID uint `json:"conversation_id,omitempty"` // required for ConversationChannel + PubsubToken string `json:"pubsub_token,omitempty"` // required for contact RoomChannel } // Channel name constants (ActionCable naming style) const ( ChannelAccount = "AccountChannel" ChannelConversation = "ConversationChannel" - ChannelRoom = "RoomChannel" // Chatwoot single-subscription channel + ChannelRoom = "RoomChannel" // Chatwoot single-subscription channel ) // --- Server → Client Frames --- diff --git a/backend/internal/handler/ws/ws_test.go b/backend/internal/handler/ws/ws_test.go index 5e61a05a..1395b0d6 100644 --- a/backend/internal/handler/ws/ws_test.go +++ b/backend/internal/handler/ws/ws_test.go @@ -1,8 +1,10 @@ package ws import ( + "bytes" "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" "strings" @@ -19,9 +21,15 @@ import ( "gorm.io/gorm" "github.com/gochat/gochat/internal/auth" + "github.com/gochat/gochat/internal/channel" "github.com/gochat/gochat/internal/config" + v1 "github.com/gochat/gochat/internal/handler/api/v1" + widgethandler "github.com/gochat/gochat/internal/handler/widget" "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/repository" + "github.com/gochat/gochat/internal/service" wspkg "github.com/gochat/gochat/internal/ws" + "github.com/gochat/gochat/internal/wsevent" ) // --- Protocol Tests --- @@ -268,6 +276,8 @@ func TestHub_SendToAccountSanitizesVisitorIdentity(t *testing.T) { visitor.IsContact = true visitor.Identifier = `{"channel":"AccountChannel","account_id":10}` hub.Register(visitor) + hub.subscribeClient(visitor.ID, accountRoomName(10)) + visitor.SubscribedRooms[accountRoomName(10)] = true hub.SendToAccount(10, []byte(`{"event":"message.created","data":{"content":"same reply","sender_type":"AgentBot","sender_id":7,"ai_takeover_active":true,"additional_attributes":{"agent_name":"Captain"}}}`)) @@ -470,6 +480,38 @@ func TestUintToStr(t *testing.T) { assert.Equal(t, "0", uintToStr(0)) } +func TestHandleSubscribe_ContactRoomUsesAuthenticatedPubsubToken(t *testing.T) { + hub := NewHubSimple() + handler := NewHandler(hub, nil) + client := NewClient(9, 10, nil, hub) + client.IsContact = true + client.PubsubToken = "visitor-token" + + identifier, err := json.Marshal(ChannelIdentifier{ + Channel: ChannelRoom, + PubsubToken: "visitor-token", + }) + require.NoError(t, err) + handler.handleSubscribe(client, CommandFrame{Command: CommandSubscribe, Identifier: string(identifier)}) + + assert.True(t, client.SubscribedRooms[pubsubTokenRoomName("visitor-token")]) + var confirm ConfirmFrame + require.NoError(t, json.Unmarshal(<-client.Send, &confirm)) + assert.Equal(t, ServerConfirmSubscribe, confirm.Type) + + other := NewClient(9, 10, nil, hub) + other.IsContact = true + other.PubsubToken = "visitor-token" + mismatch, err := json.Marshal(ChannelIdentifier{Channel: ChannelRoom, PubsubToken: "other-token"}) + require.NoError(t, err) + handler.handleSubscribe(other, CommandFrame{Command: CommandSubscribe, Identifier: string(mismatch)}) + + assert.False(t, other.SubscribedRooms[pubsubTokenRoomName("other-token")]) + var reject RejectFrame + require.NoError(t, json.Unmarshal(<-other.Send, &reject)) + assert.Equal(t, ServerRejectSubscribe, reject.Type) +} + // --- Integration: ServeWS with real WebSocket --- // createTestHandler creates a Handler with a real WSAuthenticator using a test JWT config. @@ -587,6 +629,214 @@ func TestServeWS_ValidToken_Success(t *testing.T) { assert.Equal(t, ServerPing, pingResp.Type) } +func TestServeCable_WidgetReceivesTokenRoomEvent(t *testing.T) { + gin.SetMode(gin.TestMode) + db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=private"), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.AutoMigrate(&model.Account{}, &model.Contact{}, &model.Inbox{}, &model.ContactInbox{})) + account := model.Account{Name: "Widget account"} + require.NoError(t, db.Create(&account).Error) + contact := model.Contact{AccountID: account.ID, Name: "Visitor"} + require.NoError(t, db.Create(&contact).Error) + inbox := model.Inbox{AccountID: account.ID, Name: "Website", ChannelType: "Channel::WebWidget", Enabled: true} + require.NoError(t, db.Create(&inbox).Error) + contactInbox := model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, PubsubToken: "visitor-token"} + require.NoError(t, db.Create(&contactInbox).Error) + + hub := NewHubSimple() + authenticator := wspkg.NewWSAuthenticator(nil, repository.NewContactInboxRepo(db), db) + handler := NewHandler(hub, authenticator) + router := gin.New() + router.GET("/cable", handler.ServeCable) + server := httptest.NewServer(router) + t.Cleanup(server.Close) + + conn, _, err := websocket.DefaultDialer.Dial( + "ws"+strings.TrimPrefix(server.URL, "http")+"/cable?pubsub_token=visitor-token", + nil, + ) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + require.NoError(t, conn.SetReadDeadline(time.Now().Add(2*time.Second))) + _, _, err = conn.ReadMessage() // welcome + require.NoError(t, err) + + identifier, err := json.Marshal(ChannelIdentifier{Channel: ChannelRoom, PubsubToken: "visitor-token"}) + require.NoError(t, err) + command, err := json.Marshal(CommandFrame{Command: CommandSubscribe, Identifier: string(identifier)}) + require.NoError(t, err) + require.NoError(t, conn.WriteMessage(websocket.TextMessage, command)) + _, confirmation, err := conn.ReadMessage() + require.NoError(t, err) + var confirm ConfirmFrame + require.NoError(t, json.Unmarshal(confirmation, &confirm)) + assert.Equal(t, ServerConfirmSubscribe, confirm.Type) + + hub.SendToRoom(pubsubTokenRoomName("visitor-token"), []byte(`{"event":"message.created","data":{"id":12,"content":"Dashboard reply","message_type":1,"conversation_id":42}}`)) + _, message, err := conn.ReadMessage() + require.NoError(t, err) + var delivered struct { + Identifier string `json:"identifier"` + Message json.RawMessage `json:"message"` + } + require.NoError(t, json.Unmarshal(message, &delivered)) + assert.JSONEq(t, string(identifier), delivered.Identifier) + var event wspkg.WSMessage + require.NoError(t, json.Unmarshal(delivered.Message, &event)) + assert.Equal(t, wspkg.EventMessageCreated, event.Event) + payload := event.Data.(map[string]interface{}) + assert.Equal(t, "Dashboard reply", payload["content"]) + assert.Equal(t, float64(1), payload["message_type"]) + assert.Equal(t, float64(42), payload["conversation_id"]) +} + +func TestDashboardOutgoingReachesDashboardAndReconnectedWidget(t *testing.T) { + gin.SetMode(gin.TestMode) + db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=private"), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.AutoMigrate( + &model.Account{}, &model.User{}, &model.Inbox{}, &model.Contact{}, &model.ContactInbox{}, + &model.Conversation{}, &model.Message{}, &model.Attachment{}, + )) + account := &model.Account{Name: "Realtime account", Active: true} + require.NoError(t, db.Create(account).Error) + inbox := &model.Inbox{AccountID: account.ID, Name: "Website", ChannelType: "web_widget", Enabled: true} + require.NoError(t, db.Create(inbox).Error) + contact := &model.Contact{AccountID: account.ID, Name: "Visitor"} + require.NoError(t, db.Create(contact).Error) + contactInbox := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, PubsubToken: "visitor-token"} + require.NoError(t, db.Create(contactInbox).Error) + displayID := uint(42) + conversation := &model.Conversation{ + AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, ContactInboxID: &contactInbox.ID, + DisplayID: &displayID, Status: "open", ChannelType: "web_widget", Channel: "web_widget", + } + require.NoError(t, db.Create(conversation).Error) + + hub := NewHubSimple() + dispatcher := channel.NewDispatcher() + dispatcher.Register(wsevent.New(wspkg.NewEventPublisherLocal(hub, nil))) + messageService := service.NewMessageService(repository.NewMessageRepo(db), dispatcher, nil) + messageHandler := v1.NewMessageHandler(messageService) + widgetService := service.NewWidgetService( + repository.NewInboxRepo(db), repository.NewContactRepo(db), repository.NewContactInboxRepo(db), + repository.NewConversationRepo(db), repository.NewMessageRepo(db), nil, nil, nil, nil, nil, nil, nil, nil, + ) + widgetHandler := widgethandler.NewHandler(widgetService) + jwtService := auth.NewJWTService(&config.JWTConfig{Secret: "dashboard-widget-chain", ExpiryHours: 1, AccessExpiryMinutes: 60}) + wsHandler := NewHandler(hub, wspkg.NewWSAuthenticator(jwtService, repository.NewContactInboxRepo(db))) + + router := gin.New() + router.Use(func(c *gin.Context) { + c.Set("user_id", uint(7)) + c.Next() + }) + router.GET("/cable", wsHandler.ServeCable) + router.POST("/api/v1/accounts/:account_id/conversations/:conversation_id/messages", messageHandler.Create) + router.GET("/api/v1/widget/messages", widgetHandler.GetLatestMessages) + server := httptest.NewServer(router) + t.Cleanup(server.Close) + + user := &model.User{Base: model.Base{ID: 7}, Provider: "local"} + tokenPair, err := jwtService.GenerateTokenPair(user, account.ID, "agent") + require.NoError(t, err) + dashboard := dialCable(t, server.URL, "?token="+tokenPair.AccessToken) + t.Cleanup(func() { _ = dashboard.Close() }) + dashboardIdentifier := subscribeCable(t, dashboard, ChannelIdentifier{Channel: ChannelRoom, AccountID: account.ID}) + + visitor := dialCable(t, server.URL, "?pubsub_token="+contactInbox.PubsubToken) + mismatch, err := json.Marshal(ChannelIdentifier{Channel: ChannelRoom, PubsubToken: "other-token"}) + require.NoError(t, err) + require.NoError(t, visitor.WriteJSON(CommandFrame{Command: CommandSubscribe, Identifier: string(mismatch)})) + var rejected RejectFrame + require.NoError(t, visitor.ReadJSON(&rejected)) + assert.Equal(t, ServerRejectSubscribe, rejected.Type) + visitorIdentifier := subscribeCable(t, visitor, ChannelIdentifier{Channel: ChannelRoom, PubsubToken: contactInbox.PubsubToken}) + + createDashboardMessage(t, server.URL, account.ID, displayID, "dashboard reply one") + assertCableMessage(t, dashboard, dashboardIdentifier, "dashboard reply one", displayID) + assertCableMessage(t, visitor, visitorIdentifier, "dashboard reply one", displayID) + + refreshRequest, err := http.NewRequest(http.MethodGet, server.URL+"/api/v1/widget/messages", nil) + require.NoError(t, err) + refreshRequest.Header.Set("X-Auth-Token", contactInbox.PubsubToken) + refreshResponse, err := http.DefaultClient.Do(refreshRequest) + require.NoError(t, err) + defer refreshResponse.Body.Close() + require.Equal(t, http.StatusOK, refreshResponse.StatusCode) + var refresh struct { + Payload []struct { + Content string `json:"content"` + ConversationID uint `json:"conversation_id"` + } `json:"payload"` + } + require.NoError(t, json.NewDecoder(refreshResponse.Body).Decode(&refresh)) + require.Len(t, refresh.Payload, 1) + assert.Equal(t, "dashboard reply one", refresh.Payload[0].Content) + assert.Equal(t, displayID, refresh.Payload[0].ConversationID) + + require.NoError(t, visitor.Close()) + visitor = dialCable(t, server.URL, "?pubsub_token="+contactInbox.PubsubToken) + t.Cleanup(func() { _ = visitor.Close() }) + visitorIdentifier = subscribeCable(t, visitor, ChannelIdentifier{Channel: ChannelRoom, PubsubToken: contactInbox.PubsubToken}) + createDashboardMessage(t, server.URL, account.ID, displayID, "dashboard reply after reconnect") + assertCableMessage(t, dashboard, dashboardIdentifier, "dashboard reply after reconnect", displayID) + assertCableMessage(t, visitor, visitorIdentifier, "dashboard reply after reconnect", displayID) +} + +func dialCable(t *testing.T, serverURL, query string) *websocket.Conn { + t.Helper() + conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(serverURL, "http")+"/cable"+query, nil) + require.NoError(t, err) + require.NoError(t, conn.SetReadDeadline(time.Now().Add(2*time.Second))) + var welcome WelcomeFrame + require.NoError(t, conn.ReadJSON(&welcome)) + require.Equal(t, ServerWelcome, welcome.Type) + return conn +} + +func subscribeCable(t *testing.T, conn *websocket.Conn, identifier ChannelIdentifier) string { + t.Helper() + raw, err := json.Marshal(identifier) + require.NoError(t, err) + require.NoError(t, conn.WriteJSON(CommandFrame{Command: CommandSubscribe, Identifier: string(raw)})) + var confirmed ConfirmFrame + require.NoError(t, conn.ReadJSON(&confirmed)) + require.Equal(t, ServerConfirmSubscribe, confirmed.Type) + return string(raw) +} + +func createDashboardMessage(t *testing.T, serverURL string, accountID, displayID uint, content string) { + t.Helper() + body, err := json.Marshal(map[string]any{"content": content, "message_type": "outgoing"}) + require.NoError(t, err) + request, err := http.NewRequest(http.MethodPost, + fmt.Sprintf("%s/api/v1/accounts/%d/conversations/%d/messages", serverURL, accountID, displayID), bytes.NewReader(body)) + require.NoError(t, err) + request.Header.Set("Content-Type", "application/json") + response, err := http.DefaultClient.Do(request) + require.NoError(t, err) + defer response.Body.Close() + require.Equal(t, http.StatusOK, response.StatusCode) +} + +func assertCableMessage(t *testing.T, conn *websocket.Conn, identifier, content string, displayID uint) { + t.Helper() + var delivered struct { + Identifier string `json:"identifier"` + Message json.RawMessage `json:"message"` + } + require.NoError(t, conn.ReadJSON(&delivered)) + assert.JSONEq(t, identifier, delivered.Identifier) + var event wspkg.WSMessage + require.NoError(t, json.Unmarshal(delivered.Message, &event)) + require.Equal(t, wspkg.EventMessageCreated, event.Event) + payload := event.Data.(map[string]interface{}) + assert.Equal(t, content, payload["content"]) + assert.Equal(t, float64(1), payload["message_type"]) + assert.Equal(t, float64(displayID), payload["conversation_id"]) +} + func TestRemoteSocketRechecksAccessWhenDisconnectPublishFails(t *testing.T) { gin.SetMode(gin.TestMode) db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=private"), &gorm.Config{}) diff --git a/backend/internal/repository/contact_inbox_repo.go b/backend/internal/repository/contact_inbox_repo.go index 38325548..1913c207 100644 --- a/backend/internal/repository/contact_inbox_repo.go +++ b/backend/internal/repository/contact_inbox_repo.go @@ -39,6 +39,16 @@ 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 +} + // FindByContactInboxSource retrieves a contact_inbox by the Chatwoot builder identity. func (r *ContactInboxRepo) FindByContactInboxSource(ctx context.Context, contactID, inboxID uint, sourceID string) (*model.ContactInbox, error) { var ci model.ContactInbox diff --git a/backend/internal/repository/conversation_repo.go b/backend/internal/repository/conversation_repo.go index 514b171c..4b649b64 100644 --- a/backend/internal/repository/conversation_repo.go +++ b/backend/internal/repository/conversation_repo.go @@ -190,6 +190,22 @@ func (r *ConversationRepo) FindByContact(ctx context.Context, accountID, contact return conversations, total, err } +// FindByContactInbox retrieves conversations belonging to one channel identity. +// A contact may have several inbox identities, but a widget token must only see +// the conversations created through its own ContactInbox. +func (r *ConversationRepo) FindByContactInbox(ctx context.Context, accountID, contactID, inboxID, contactInboxID uint, includeLegacy bool, offset, limit int) ([]model.Conversation, int64, error) { + var conversations []model.Conversation + var total int64 + query := r.db.WithContext(ctx).Model(&model.Conversation{}). + Where("account_id = ? AND (contact_inbox_id = ? OR (contact_inbox_id IS NULL AND contact_id = ? AND inbox_id = ? AND ?))", + accountID, contactInboxID, contactID, inboxID, includeLegacy) + if err := query.Count(&total).Error; err != nil { + return nil, 0, err + } + err := query.Offset(offset).Limit(limit).Order("id DESC").Find(&conversations).Error + return conversations, total, err +} + // FindRecentByContact retrieves the latest conversations for a contact. // Reference: Chatwoot contacts/conversations#index limits to the latest 20 conversations ordered by last_activity_at. func (r *ConversationRepo) FindRecentByContact(ctx context.Context, accountID, contactID uint, inboxID *uint, limit int) ([]model.Conversation, error) { diff --git a/backend/internal/repository/conversation_repo_test.go b/backend/internal/repository/conversation_repo_test.go index 836fb0af..f72672d6 100644 --- a/backend/internal/repository/conversation_repo_test.go +++ b/backend/internal/repository/conversation_repo_test.go @@ -58,6 +58,33 @@ func TestConversationRepo_FindByID_NotFound(t *testing.T) { assert.Nil(t, found) } +func TestConversationRepo_FindByContactInboxIncludesLegacyOnlyWhenAllowed(t *testing.T) { + db := setupTestDB(t) + repo := NewConversationRepo(db) + account := &model.Account{Name: "Widget identity account", Active: true} + require.NoError(t, db.Create(account).Error) + inbox := &model.Inbox{AccountID: account.ID, Name: "Widget", ChannelType: "web_widget"} + require.NoError(t, db.Create(inbox).Error) + contact := &model.Contact{AccountID: account.ID, Name: "Visitor"} + require.NoError(t, db.Create(contact).Error) + contactInbox := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, PubsubToken: "token-a"} + require.NoError(t, db.Create(contactInbox).Error) + legacy := createTestConversation(t, db, account.ID, inbox.ID, contact.ID, "open") + owned := createTestConversation(t, db, account.ID, inbox.ID, contact.ID, "open") + require.NoError(t, db.Model(owned).Update("contact_inbox_id", contactInbox.ID).Error) + + conversations, total, err := repo.FindByContactInbox(context.Background(), account.ID, contact.ID, inbox.ID, contactInbox.ID, true, 0, 10) + require.NoError(t, err) + assert.Equal(t, int64(2), total) + assert.ElementsMatch(t, []uint{legacy.ID, owned.ID}, []uint{conversations[0].ID, conversations[1].ID}) + + conversations, total, err = repo.FindByContactInbox(context.Background(), account.ID, contact.ID, inbox.ID, contactInbox.ID, false, 0, 10) + require.NoError(t, err) + assert.Equal(t, int64(1), total) + require.Len(t, conversations, 1) + assert.Equal(t, owned.ID, conversations[0].ID) +} + func TestConversationRepo_FindByAccountAndID(t *testing.T) { db := setupTestDB(t) repo := NewConversationRepo(db) diff --git a/backend/internal/service/widget_service.go b/backend/internal/service/widget_service.go index d48262b6..6219f42e 100644 --- a/backend/internal/service/widget_service.go +++ b/backend/internal/service/widget_service.go @@ -29,6 +29,7 @@ var ( ErrWidgetConversationNotFound = errors.New("conversation not found") ErrWidgetEndConversationDisabled = errors.New("end conversation is not permitted") ErrWidgetMessageContentTooLong = errors.New("Content is too long (maximum is 150000 characters)") + errWidgetConversationOwnership = errors.New("conversation does not belong to this contact") ) const widgetMessageContentLimit = 150000 @@ -368,16 +369,12 @@ func (s *WidgetService) SendMessage(ctx context.Context, req WidgetSendMessageRe var conversation *model.Conversation conversationCreated := false if req.ConversationID != nil { - conversation, err = s.conversationRepo.FindByID(ctx, *req.ConversationID) + conversation, err = s.resolveWidgetConversation(ctx, contactInbox, *req.ConversationID) if err != nil { return nil, fmt.Errorf("conversation not found: %w", err) } - // Verify conversation belongs to this contact - if conversation.ContactID != contactInbox.ContactID { - return nil, errors.New("conversation does not belong to this contact") - } } else { - conversations, _, findErr := s.conversationRepo.FindByContact(ctx, contactInbox.Contact.AccountID, contactInbox.ContactID, 0, 1) + conversations, _, findErr := s.findWidgetConversations(ctx, contactInbox, 0, 1) if findErr != nil { return nil, findErr } @@ -523,8 +520,7 @@ func (s *WidgetService) GetConversations(ctx context.Context, widgetToken string return nil, fmt.Errorf("invalid widget_token: %w", err) } - conversations, _, err := s.conversationRepo.FindByContact( - ctx, contactInbox.Contact.AccountID, contactInbox.ContactID, 0, 50) + conversations, _, err := s.findWidgetConversations(ctx, contactInbox, 0, 50) if err != nil { return nil, err } @@ -536,8 +532,7 @@ func (s *WidgetService) GetLatestConversation(ctx context.Context, widgetToken s if err != nil { return nil, fmt.Errorf("invalid widget_token: %w", err) } - conversations, _, err := s.conversationRepo.FindByContact( - ctx, contactInbox.Contact.AccountID, contactInbox.ContactID, 0, 1) + conversations, _, err := s.findWidgetConversations(ctx, contactInbox, 0, 1) if err != nil { return nil, err } @@ -552,12 +547,38 @@ func (s *WidgetService) GetConversation(ctx context.Context, widgetToken string, if err != nil { return nil, fmt.Errorf("invalid widget_token: %w", err) } + return s.resolveWidgetConversation(ctx, contactInbox, conversationID) +} + +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 { + return nil, 0, err + } + return s.conversationRepo.FindByContactInbox(ctx, contactInbox.Contact.AccountID, contactInbox.ContactID, + contactInbox.InboxID, contactInbox.ID, count == 1, offset, limit) +} + +func (s *WidgetService) resolveWidgetConversation(ctx context.Context, contactInbox *model.ContactInbox, conversationID uint) (*model.Conversation, error) { conversation, err := s.conversationRepo.FindByID(ctx, conversationID) if err != nil { return nil, err } - if conversation.ContactID != contactInbox.ContactID { - return nil, errors.New("conversation does not belong to this contact") + 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) + if err != nil { + return nil, err + } + if count != 1 { + return nil, errWidgetConversationOwnership } return conversation, nil } @@ -1129,14 +1150,9 @@ func (s *WidgetService) GetMessages(ctx context.Context, widgetToken string, con return nil, 0, fmt.Errorf("invalid widget_token: %w", err) } - // Verify conversation belongs to this contact - conversation, err := s.conversationRepo.FindByID(ctx, conversationID) - if err != nil { + if _, err := s.resolveWidgetConversation(ctx, contactInbox, conversationID); err != nil { return nil, 0, err } - if conversation.ContactID != contactInbox.ContactID { - return nil, 0, errors.New("conversation does not belong to this contact") - } return s.messageRepo.FindByConversation(ctx, conversationID, offset, limit) } @@ -1285,13 +1301,10 @@ func (s *WidgetService) UpdateMessage(ctx context.Context, req WidgetMessageUpda if err != nil { return nil, nil, err } - conversation, err := s.conversationRepo.FindByID(ctx, message.ConversationID) + conversation, err := s.resolveWidgetConversation(ctx, contactInbox, message.ConversationID) if err != nil { return nil, nil, err } - if conversation.ContactID != contactInbox.ContactID || conversation.InboxID != contactInbox.InboxID { - return nil, nil, errors.New("message does not belong to this contact") - } contact := &contactInbox.Contact if strings.TrimSpace(req.ContactEmail) != "" && message.ContentType == string(model.MessageContentTypeInputEmail) { @@ -1445,13 +1458,9 @@ func (s *WidgetService) AddDyteParticipant(ctx context.Context, websiteToken, wi if message.InboxID != inbox.ID { return nil, errors.New("message does not belong to this inbox") } - conversation, err := s.conversationRepo.FindByID(ctx, message.ConversationID) - if err != nil { + if _, err := s.resolveWidgetConversation(ctx, contactInbox, message.ConversationID); err != nil { return nil, err } - if conversation.ContactID != contactInbox.ContactID { - return nil, errors.New("message does not belong to this contact") - } if message.ContentType != "integrations" { return nil, errors.New("Invalid message type. Action not permitted") } @@ -1488,16 +1497,14 @@ func (s *WidgetService) ToggleTyping(ctx context.Context, widgetToken string, co return fmt.Errorf("invalid widget_token: %w", err) } - conversation, err := s.conversationRepo.FindByID(ctx, conversationID) + conversation, err := s.resolveWidgetConversation(ctx, contactInbox, conversationID) if err != nil { + if errors.Is(err, errWidgetConversationOwnership) { + return err + } return fmt.Errorf("conversation not found: %w", err) } - // Verify conversation belongs to this contact - if conversation.ContactID != contactInbox.ContactID { - return errors.New("conversation does not belong to this contact") - } - performer := &ws.Performer{ ID: contactInbox.ContactID, Name: contactInbox.Contact.Name, diff --git a/backend/internal/service/widget_service_test.go b/backend/internal/service/widget_service_test.go index bc2cc57e..04e634b4 100644 --- a/backend/internal/service/widget_service_test.go +++ b/backend/internal/service/widget_service_test.go @@ -787,6 +787,49 @@ func TestWidgetService_GetConversations_InvalidToken(t *testing.T) { assert.Error(t, err) } +func TestWidgetService_GetLatestMessagesScopesByContactInbox(t *testing.T) { + db, svc := setupWidgetServiceTest(t) + ctx := context.Background() + seedWidgetInbox(t, db) + initResp, err := svc.Init(ctx, WidgetInitRequest{WebsiteToken: "test_ws_token_123"}) + require.NoError(t, err) + owned, err := svc.SendMessage(ctx, WidgetSendMessageRequest{WidgetToken: initResp.WidgetToken, Content: "owned message"}) + require.NoError(t, err) + + otherContactInbox := model.ContactInbox{ + ContactID: initResp.ContactID, + InboxID: initResp.InboxID, + SourceID: "other-source", + PubsubToken: "other-widget-token", + } + require.NoError(t, db.Create(&otherContactInbox).Error) + otherConversation := model.Conversation{ + AccountID: initResp.AccountID, + InboxID: initResp.InboxID, + ContactID: initResp.ContactID, + ContactInboxID: &otherContactInbox.ID, + Status: string(model.ConversationStatusOpen), + ChannelType: string(channel.ChannelWebWidget), + Channel: "web_widget", + } + require.NoError(t, db.Create(&otherConversation).Error) + require.NoError(t, db.Create(&model.Message{ + ConversationID: otherConversation.ID, + AccountID: initResp.AccountID, + InboxID: initResp.InboxID, + Content: "other identity message", + ContentType: "text", + MessageType: string(model.MessageTypeOutgoing), + }).Error) + + messages, _, conversation, err := svc.GetLatestConversationMessages(ctx, initResp.WidgetToken, 0, 0) + require.NoError(t, err) + require.NotNil(t, conversation) + assert.Equal(t, owned.ConversationID, conversation.ID) + require.Len(t, messages, 1) + assert.Equal(t, "owned message", messages[0].Content) +} + func TestWidgetService_UpdateLastSeenQueuesMessageStatusJob(t *testing.T) { db, svc := setupWidgetServiceTest(t) ctx := context.Background() @@ -892,6 +935,96 @@ func TestWidgetService_GetMessages_WrongConversationOwner(t *testing.T) { assert.Contains(t, err.Error(), "does not belong to this contact") } +func TestWidgetService_LegacyConversationRequiresUniqueContactInbox(t *testing.T) { + db, svc := setupWidgetServiceTest(t) + ctx := context.Background() + account, inbox := seedWidgetInbox(t, db) + contact := &model.Contact{AccountID: account.ID, Name: "Legacy visitor"} + require.NoError(t, db.Create(contact).Error) + contactInbox := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, PubsubToken: "legacy-token"} + require.NoError(t, db.Create(contactInbox).Error) + legacy := &model.Conversation{ + AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, + Status: "open", ChannelType: "web_widget", Channel: "web_widget", + } + require.NoError(t, db.Create(legacy).Error) + require.NoError(t, db.Create(&model.Message{ + AccountID: account.ID, InboxID: inbox.ID, ConversationID: legacy.ID, + Content: "legacy reply", ContentType: "text", MessageType: "outgoing", Status: "sent", + }).Error) + + conversations, err := svc.GetConversations(ctx, contactInbox.PubsubToken) + require.NoError(t, err) + require.Len(t, conversations, 1) + assert.Equal(t, legacy.ID, conversations[0].ID) + _, err = svc.GetConversation(ctx, contactInbox.PubsubToken, legacy.ID) + require.NoError(t, err) + messages, _, err := svc.GetMessages(ctx, contactInbox.PubsubToken, legacy.ID, 0, 10) + require.NoError(t, err) + require.Len(t, messages, 1) + conversationID := legacy.ID + _, err = svc.SendMessage(ctx, WidgetSendMessageRequest{ + WidgetToken: contactInbox.PubsubToken, ConversationID: &conversationID, Content: "legacy visitor reply", + }) + require.NoError(t, err) + + require.NoError(t, db.Create(&model.ContactInbox{ + ContactID: contact.ID, InboxID: inbox.ID, PubsubToken: "second-legacy-token", + }).Error) + conversations, err = svc.GetConversations(ctx, contactInbox.PubsubToken) + require.NoError(t, err) + assert.Empty(t, conversations) + _, err = svc.GetConversation(ctx, contactInbox.PubsubToken, legacy.ID) + assert.ErrorIs(t, err, errWidgetConversationOwnership) + _, _, err = svc.GetMessages(ctx, contactInbox.PubsubToken, legacy.ID, 0, 10) + assert.ErrorIs(t, err, errWidgetConversationOwnership) + _, err = svc.SendMessage(ctx, WidgetSendMessageRequest{ + WidgetToken: contactInbox.PubsubToken, ConversationID: &conversationID, Content: "must fail closed", + }) + assert.ErrorIs(t, err, errWidgetConversationOwnership) +} + +func TestWidgetService_ExplicitConversationEntrypointsRejectSiblingToken(t *testing.T) { + db, svc := setupWidgetServiceTest(t) + ctx := context.Background() + account, inbox := seedWidgetInbox(t, db) + contact := &model.Contact{AccountID: account.ID, Name: "Shared contact"} + require.NoError(t, db.Create(contact).Error) + owner := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, PubsubToken: "owner-token"} + sibling := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, PubsubToken: "sibling-token"} + require.NoError(t, db.Create(owner).Error) + require.NoError(t, db.Create(sibling).Error) + conversation := &model.Conversation{ + AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, ContactInboxID: &owner.ID, + Status: "open", ChannelType: "web_widget", Channel: "web_widget", + } + require.NoError(t, db.Create(conversation).Error) + message := &model.Message{ + AccountID: account.ID, InboxID: inbox.ID, ConversationID: conversation.ID, + Content: "owner only", ContentType: "integrations", MessageType: "outgoing", Status: "sent", + ContentAttributes: mustJSON(map[string]any{"data": map[string]any{"meeting_id": "meeting-1"}}), + } + require.NoError(t, db.Create(message).Error) + + _, err := svc.GetConversation(ctx, sibling.PubsubToken, conversation.ID) + assert.ErrorIs(t, err, errWidgetConversationOwnership) + _, _, err = svc.GetMessages(ctx, sibling.PubsubToken, conversation.ID, 0, 10) + assert.ErrorIs(t, err, errWidgetConversationOwnership) + err = svc.ToggleTyping(ctx, sibling.PubsubToken, conversation.ID, true) + assert.ErrorIs(t, err, errWidgetConversationOwnership) + _, _, err = svc.UpdateMessage(ctx, WidgetMessageUpdate{ + WidgetToken: sibling.PubsubToken, MessageID: message.ID, SubmittedValues: []map[string]any{{"value": "nope"}}, + }) + assert.ErrorIs(t, err, errWidgetConversationOwnership) + _, err = svc.AddDyteParticipant(ctx, "test_ws_token_123", sibling.PubsubToken, message.ID) + assert.ErrorIs(t, err, errWidgetConversationOwnership) + conversationID := conversation.ID + _, err = svc.SendMessage(ctx, WidgetSendMessageRequest{ + WidgetToken: sibling.PubsubToken, ConversationID: &conversationID, Content: "nope", + }) + assert.ErrorIs(t, err, errWidgetConversationOwnership) +} + // ========== GetCableToken Tests ========== func TestWidgetService_GetCableToken(t *testing.T) { diff --git a/backend/internal/ws/auth.go b/backend/internal/ws/auth.go index a8fdeddc..665e3485 100644 --- a/backend/internal/ws/auth.go +++ b/backend/internal/ws/auth.go @@ -60,8 +60,8 @@ func NewWSAuthenticator(jwtService *auth.JWTService, contactInboxRepo *repositor // 1. Agent/User auth (primary): JWT token from 'token' query param or Authorization header. // Validates via jwtService.ValidateAccessToken, populates WSClaims from auth.Claims. // -// 2. Contact auth (secondary): pubsub_token + user_id query params. -// Looks up ContactInbox by pubsub_token, verifies the contact belongs to the account, +// 2. Contact auth (secondary): pubsub_token query param. +// Looks up ContactInbox by pubsub_token and resolves the contact and account, // populates WSClaims with contact identity. // // Returns WSClaims on success, or an error suitable for HTTP 401 rejection. @@ -98,18 +98,16 @@ func (a *WSAuthenticator) Authenticate(c *gin.Context) (*WSClaims, error) { // --- Path 2: Contact authentication via pubsub_token --- pubsubToken := c.Query("pubsub_token") if pubsubToken == "" { - return nil, errors.New("authentication required: provide 'token' (JWT) or 'pubsub_token' + 'user_id' params") + return nil, errors.New("authentication required: provide 'token' (JWT) or 'pubsub_token'") } - - // Contact auth requires user_id param (Chatwoot RoomChannel: contact_id from params) - userIDStr := c.Query("user_id") - if userIDStr == "" { - return nil, errors.New("contact auth requires 'user_id' parameter alongside 'pubsub_token'") - } - - contactID, err := strconv.ParseUint(userIDStr, 10, 32) - if err != nil { - return nil, fmt.Errorf("invalid user_id parameter: %w", err) + var providedContactID *uint + if userIDStr := c.Query("user_id"); userIDStr != "" { + contactID, parseErr := strconv.ParseUint(userIDStr, 10, 32) + if parseErr != nil { + return nil, fmt.Errorf("invalid user_id parameter: %w", parseErr) + } + value := uint(contactID) + providedContactID = &value } // Lookup ContactInbox by pubsub_token @@ -119,14 +117,17 @@ func (a *WSAuthenticator) Authenticate(c *gin.Context) (*WSClaims, error) { return nil, fmt.Errorf("invalid pubsub_token: %w", err) } - // Verify the contact ID matches - if contactInbox.ContactID != uint(contactID) { - logger.L().Debugf("ws auth: contact mismatch (expected=%d, found=%d)", uint(contactID), contactInbox.ContactID) - return nil, errors.New("pubsub_token does not belong to the specified contact") + // Legacy clients may still send user_id. Treat it as an additional + // fail-closed check, while Chatwoot widgets authenticate by token alone. + if providedContactID != nil { + if contactInbox.ContactID != *providedContactID { + logger.L().Debugf("ws auth: contact mismatch (expected=%d, found=%d)", *providedContactID, contactInbox.ContactID) + return nil, errors.New("pubsub_token does not belong to the specified contact") + } } wsClaims := &WSClaims{ - UserID: uint(contactID), // for contacts, UserID maps to contact_id (Chatwoot convention) + UserID: contactInbox.ContactID, // for contacts, UserID maps to contact_id (Chatwoot convention) AccountID: contactInbox.Contact.AccountID, Role: "contact", Provider: "pubsub_token", @@ -208,6 +209,9 @@ func (a *WSAuthenticator) AuthenticateAndServeWS(c *gin.Context) { // findContactInboxByPubsubToken looks up a ContactInbox by its PubsubToken field. // Preloads the associated Contact to resolve the AccountID for authorization. func (a *WSAuthenticator) findContactInboxByPubsubToken(ctx context.Context, pubsubToken string) (*model.ContactInbox, error) { + if a.contactInboxRepo == nil { + return nil, errors.New("contact inbox repository unavailable") + } return a.contactInboxRepo.FindByPubsubToken(ctx, pubsubToken) } diff --git a/backend/internal/ws/coverage4_test.go b/backend/internal/ws/coverage4_test.go index 00b9a097..58f141ed 100644 --- a/backend/internal/ws/coverage4_test.go +++ b/backend/internal/ws/coverage4_test.go @@ -36,7 +36,7 @@ func TestWSAuthenticator_Authenticate_PubsubTokenNoUserID_Cov4(t *testing.T) { a := NewWSAuthenticator(nil, nil) _, err := a.Authenticate(c) assert.Error(t, err) - assert.Contains(t, err.Error(), "user_id") + assert.Contains(t, err.Error(), "invalid pubsub_token") } func TestWSAuthenticator_Authenticate_PubsubTokenInvalidUserID_Cov4(t *testing.T) { diff --git a/backend/internal/ws/coverage5_test.go b/backend/internal/ws/coverage5_test.go index 21ee56be..eca9486b 100644 --- a/backend/internal/ws/coverage5_test.go +++ b/backend/internal/ws/coverage5_test.go @@ -109,7 +109,7 @@ func TestAuthenticate_PubsubTokenWithoutUserID_Cov5(t *testing.T) { claims, err := authenticator.Authenticate(c) require.Error(t, err) assert.Nil(t, claims) - assert.Contains(t, err.Error(), "requires 'user_id' parameter") + assert.Contains(t, err.Error(), "invalid pubsub_token") } func TestAuthenticate_PubsubToken_InvalidUserID_Cov5(t *testing.T) { diff --git a/backend/internal/ws/coverage7_test.go b/backend/internal/ws/coverage7_test.go index 47f641cf..2de50bad 100644 --- a/backend/internal/ws/coverage7_test.go +++ b/backend/internal/ws/coverage7_test.go @@ -142,7 +142,7 @@ func TestAuthenticate_PubsubNoUserID_Cov7(t *testing.T) { claims, err := a.Authenticate(c) require.Error(t, err) assert.Nil(t, claims) - assert.Contains(t, err.Error(), "user_id") + assert.Contains(t, err.Error(), "invalid pubsub_token") } func TestAuthenticate_PubsubInvalidUserID_Cov7(t *testing.T) { diff --git a/backend/internal/wsevent/bridge_listener_test.go b/backend/internal/wsevent/bridge_listener_test.go index 890f7f69..e391d5dd 100644 --- a/backend/internal/wsevent/bridge_listener_test.go +++ b/backend/internal/wsevent/bridge_listener_test.go @@ -31,7 +31,9 @@ func (h *captureHub) SendToRoom(room string, data []byte) { func TestBridgeListenerRoutesWebWidgetEventsToDashboardAndVisitor(t *testing.T) { event := channel.NewChannelEvent(channel.EventMessageCreated, channel.ChannelWebWidget, 1, 4) event.Data["widget_token"] = "visitor-token" + displayID := uint(42) event.Data["message"] = &model.Message{Base: model.Base{ID: 12}, AccountID: 1, InboxID: 4, ConversationID: 2, MessageType: "outgoing", SenderType: "User"} + event.Data["conversation"] = &model.Conversation{Base: model.Base{ID: 2}, DisplayID: &displayID} event.Data["sender"] = &model.User{Base: model.Base{ID: 8}, Name: "Agent A"} hub := &captureHub{} @@ -53,6 +55,9 @@ func TestBridgeListenerRoutesWebWidgetEventsToDashboardAndVisitor(t *testing.T) if payload["sender_type"] != "User" || payload["sender"].(map[string]interface{})["name"] != "Agent A" { t.Fatalf("visitor reply sender contract mismatch: %#v", payload) } + if payload["message_type"] != float64(1) || payload["conversation_id"] != float64(displayID) { + t.Fatalf("visitor reply message contract mismatch: %#v", payload) + } } func TestBridgeListenerPreservesWebWidgetSenderTypeContract(t *testing.T) { diff --git a/backend/migrations/000082_backfill_widget_conversation_contact_inboxes.down.sql b/backend/migrations/000082_backfill_widget_conversation_contact_inboxes.down.sql new file mode 100644 index 00000000..d7f95246 --- /dev/null +++ b/backend/migrations/000082_backfill_widget_conversation_contact_inboxes.down.sql @@ -0,0 +1,10 @@ +UPDATE conversations +SET contact_inbox_id = NULL +WHERE EXISTS ( + SELECT 1 + FROM widget_conversation_contact_inbox_backfills AS backfills + WHERE backfills.conversation_id = conversations.id + AND backfills.contact_inbox_id = conversations.contact_inbox_id +); + +DROP TABLE widget_conversation_contact_inbox_backfills; diff --git a/backend/migrations/000082_backfill_widget_conversation_contact_inboxes.up.sql b/backend/migrations/000082_backfill_widget_conversation_contact_inboxes.up.sql new file mode 100644 index 00000000..4a16bd8a --- /dev/null +++ b/backend/migrations/000082_backfill_widget_conversation_contact_inboxes.up.sql @@ -0,0 +1,24 @@ +CREATE TABLE widget_conversation_contact_inbox_backfills ( + conversation_id BIGINT PRIMARY KEY, + contact_inbox_id BIGINT NOT NULL +); + +INSERT INTO widget_conversation_contact_inbox_backfills (conversation_id, contact_inbox_id) +SELECT conversations.id, MIN(contact_inboxes.id) +FROM conversations +JOIN inboxes ON inboxes.id = conversations.inbox_id +JOIN contact_inboxes + ON contact_inboxes.contact_id = conversations.contact_id + AND contact_inboxes.inbox_id = conversations.inbox_id +WHERE conversations.contact_inbox_id IS NULL + AND inboxes.channel_type IN ('web_widget', 'Channel::WebWidget') +GROUP BY conversations.id +HAVING COUNT(contact_inboxes.id) = 1; + +UPDATE conversations +SET contact_inbox_id = ( + SELECT backfills.contact_inbox_id + FROM widget_conversation_contact_inbox_backfills AS backfills + WHERE backfills.conversation_id = conversations.id +) +WHERE id IN (SELECT conversation_id FROM widget_conversation_contact_inbox_backfills); diff --git a/frontend/app/javascript/shared/helpers/BaseActionCableConnector.js b/frontend/app/javascript/shared/helpers/BaseActionCableConnector.js index 370adcb2..b5ead52b 100644 --- a/frontend/app/javascript/shared/helpers/BaseActionCableConnector.js +++ b/frontend/app/javascript/shared/helpers/BaseActionCableConnector.js @@ -30,6 +30,8 @@ class BaseActionCableConnector { let websocketURL = `${wsOrigin}/cable`; if (accessToken) { websocketURL += `?access-token=${encodeURIComponent(accessToken)}`; + } else if (pubsubToken) { + websocketURL += `?pubsub_token=${encodeURIComponent(pubsubToken)}`; } this.consumer = createConsumer(websocketURL); diff --git a/frontend/app/javascript/shared/helpers/specs/BaseActionCableConnector.spec.js b/frontend/app/javascript/shared/helpers/specs/BaseActionCableConnector.spec.js new file mode 100644 index 00000000..da41e50d --- /dev/null +++ b/frontend/app/javascript/shared/helpers/specs/BaseActionCableConnector.spec.js @@ -0,0 +1,93 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; + +const { createConsumer, createSubscription, getCookie, subscriptionState } = + vi.hoisted(() => { + const subscriptionState = { callbacks: null }; + return { + createConsumer: vi.fn(), + createSubscription: vi.fn((_identifier, callbacks) => { + subscriptionState.callbacks = callbacks; + return { updatePresence: vi.fn() }; + }), + getCookie: vi.fn(), + subscriptionState, + }; + }); + +vi.mock('@rails/actioncable', () => ({ createConsumer })); +vi.mock('js-cookie', () => ({ default: { get: getCookie } })); + +import BaseActionCableConnector from '../BaseActionCableConnector'; + +describe('BaseActionCableConnector', () => { + afterEach(() => { + vi.clearAllMocks(); + vi.useRealTimers(); + subscriptionState.callbacks = null; + BaseActionCableConnector.isDisconnected = false; + }); + + it('authenticates a widget socket and RoomChannel with its pubsub token', () => { + vi.useFakeTimers(); + getCookie.mockReturnValue(undefined); + createConsumer.mockReturnValue({ + subscriptions: { create: createSubscription }, + }); + + new BaseActionCableConnector( + { $store: { getters: {} } }, + 'visitor token' + ); + + expect(createConsumer).toHaveBeenCalledWith( + `${window.location.origin}/cable?pubsub_token=visitor%20token` + ); + expect(createSubscription).toHaveBeenCalledWith( + expect.objectContaining({ + channel: 'RoomChannel', + pubsub_token: 'visitor token', + }), + expect.any(Object) + ); + }); + + it('continues consuming widget events after ActionCable reconnects', () => { + vi.useFakeTimers(); + getCookie.mockReturnValue(undefined); + const isOpen = vi.fn(() => true); + createConsumer.mockReturnValue({ + connection: { isOpen }, + subscriptions: { create: createSubscription }, + }); + const connector = new BaseActionCableConnector( + { $store: { getters: {} } }, + 'visitor-token' + ); + const onMessage = vi.fn(); + connector.events['message.created'] = onMessage; + connector.onDisconnected = vi.fn(); + connector.onReconnect = vi.fn(); + + subscriptionState.callbacks.received({ + event: 'message.created', + data: { id: 1, content: 'before reconnect' }, + }); + subscriptionState.callbacks.disconnected(); + vi.advanceTimersByTime(1000); + subscriptionState.callbacks.received({ + event: 'message.created', + data: { id: 2, content: 'after reconnect' }, + }); + + expect(connector.onDisconnected).toHaveBeenCalledOnce(); + expect(connector.onReconnect).toHaveBeenCalledOnce(); + expect(onMessage).toHaveBeenNthCalledWith(1, { + id: 1, + content: 'before reconnect', + }); + expect(onMessage).toHaveBeenNthCalledWith(2, { + id: 2, + content: 'after reconnect', + }); + }); +});