diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6d07b7f7..cd666931 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -263,7 +263,7 @@ jobs: run: | set -euo pipefail mkdir -p "$GOCHAT_WS_E2E_EVIDENCE_DIR" - go test -json -count=1 -timeout 2m ./tests/e2e -run '^TestWebSocketMultiInstanceFanout$' \ + go test -json -count=3 -timeout 6m ./tests/e2e -run '^TestWebSocketMultiInstanceFanout$' \ | tee "$GOCHAT_WS_E2E_EVIDENCE_DIR/test.jsonl" grep -Eq '"Action":"pass".*"Test":"TestWebSocketMultiInstanceFanout"' \ "$GOCHAT_WS_E2E_EVIDENCE_DIR/test.jsonl" diff --git a/backend/cmd/gochat/main.go b/backend/cmd/gochat/main.go index ec195348..966eb2f5 100644 --- a/backend/cmd/gochat/main.go +++ b/backend/cmd/gochat/main.go @@ -153,6 +153,7 @@ type smokeSeedSummary struct { ConversationID uint `json:"conversation_id"` ConversationDisplayID uint `json:"conversation_display_id"` ConversationUID string `json:"conversation_uuid"` + WidgetToken string `json:"widget_token"` CsatMessageID uint `json:"csat_message_id"` SlaPolicyID uint `json:"sla_policy_id"` CustomRoleID uint `json:"custom_role_id"` @@ -266,7 +267,8 @@ func seedSmokeDataInTransaction(ctx context.Context, db *gorm.DB) (*smokeSeedSum return nil, fmt.Errorf("update contact company: %w", err) } contactInbox := &model.ContactInbox{} - if err := db.WithContext(ctx).Where("contact_id = ? AND inbox_id = ?", contact.ID, inbox.ID).FirstOrCreate(contactInbox, model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, SourceID: "gochat-smoke-source", PubsubToken: "gochat-smoke-contact-pubsub"}).Error; err != nil { + widgetToken := fmt.Sprintf("gochat-smoke-contact-pubsub-%d-%d", account.ID, inbox.ID) + if err := db.WithContext(ctx).Where("contact_id = ? AND inbox_id = ?", contact.ID, inbox.ID).FirstOrCreate(contactInbox, model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, SourceID: "gochat-smoke-source", PubsubToken: widgetToken}).Error; err != nil { return nil, fmt.Errorf("seed contact inbox: %w", err) } @@ -333,7 +335,7 @@ func seedSmokeDataInTransaction(ctx context.Context, db *gorm.DB) (*smokeSeedSum if conversation.DisplayID != nil { conversationDisplayID = *conversation.DisplayID } - return &smokeSeedSummary{AdminID: admin.ID, AdminEmail: adminEmail, AdminPassword: adminPassword, AccountID: account.ID, InboxID: inbox.ID, VoiceInboxID: voiceInbox.ID, ContactID: contact.ID, CompanyID: company.ID, PortalID: portal.ID, ArticleID: article.ID, ConversationID: conversation.ID, ConversationDisplayID: conversationDisplayID, ConversationUID: conversation.UUID, CsatMessageID: csatMessage.ID, SlaPolicyID: sla.ID, CustomRoleID: customRole.ID, CapacityPolicyID: capacity.ID, CaptainAssistantID: assistant.ID, CaptainMessageID: captainMessage.ID, AgentBotID: agentBot.ID}, nil + return &smokeSeedSummary{AdminID: admin.ID, AdminEmail: adminEmail, AdminPassword: adminPassword, AccountID: account.ID, InboxID: inbox.ID, VoiceInboxID: voiceInbox.ID, ContactID: contact.ID, CompanyID: company.ID, PortalID: portal.ID, ArticleID: article.ID, ConversationID: conversation.ID, ConversationDisplayID: conversationDisplayID, ConversationUID: conversation.UUID, WidgetToken: contactInbox.PubsubToken, CsatMessageID: csatMessage.ID, SlaPolicyID: sla.ID, CustomRoleID: customRole.ID, CapacityPolicyID: capacity.ID, CaptainAssistantID: assistant.ID, CaptainMessageID: captainMessage.ID, AgentBotID: agentBot.ID}, nil } func seedSmokeCaptain(ctx context.Context, db *gorm.DB, accountID, inboxID uint) (*model.CaptainAssistant, error) { diff --git a/backend/internal/handler/api/v1/contact_handler_crud_test.go b/backend/internal/handler/api/v1/contact_handler_crud_test.go index 55901d0c..f3eb048f 100644 --- a/backend/internal/handler/api/v1/contact_handler_crud_test.go +++ b/backend/internal/handler/api/v1/contact_handler_crud_test.go @@ -1227,7 +1227,8 @@ func (s *ContactHandlerCRUDTestSuite) TestListAttachmentsTimelineDepthMatchesCha firstSender := first["sender"].(map[string]any) s.Equal(float64(s.user.ID), firstSender["id"]) s.Equal("CRUDTestUser", firstSender["name"]) - s.Equal("agent", firstSender["role"]) + s.Equal("user", firstSender["type"]) + s.NotContains(firstSender, "role") s.Equal(float64(newMessage.CreatedAt.Unix()), first["created_at"]) second := payload[1].(map[string]any) diff --git a/backend/internal/handler/api/v1/conversation_serializer.go b/backend/internal/handler/api/v1/conversation_serializer.go index 2d0ce8d5..077580d7 100644 --- a/backend/internal/handler/api/v1/conversation_serializer.go +++ b/backend/internal/handler/api/v1/conversation_serializer.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "os" - "path/filepath" "strings" "time" @@ -457,22 +456,22 @@ func serializeMessage(ctx context.Context, db *gorm.DB, message *model.Message, case "contact": var contact model.Contact if err := db.WithContext(ctx).First(&contact, *message.SenderID).Error; err == nil { - payload.Sender = serializeContactWithContext(ctx, &contact) + payload.Sender = contact.PushEventData() } case "agent_bot": var bot model.AgentBot if err := db.WithContext(ctx).First(&bot, *message.SenderID).Error; err == nil { - payload.Sender = serializeAgentBotSender(&bot) + payload.Sender = bot.PushEventData() } case "captain_assistant": var assistant model.CaptainAssistant if err := db.WithContext(ctx).First(&assistant, *message.SenderID).Error; err == nil { - payload.Sender = serializeCaptainAssistantSender(&assistant) + payload.Sender = assistant.PushEventData() } default: var user model.User if err := db.WithContext(ctx).First(&user, *message.SenderID).Error; err == nil { - payload.Sender = serializeUser(&user, message.AccountID) + payload.Sender = user.PushEventData() } } } @@ -616,27 +615,7 @@ func serializeAttachment(ctx context.Context, db *gorm.DB, attachment *model.Att } func serializeAttachmentPushEventData(attachment *model.Attachment) map[string]any { - extension := strings.TrimPrefix(filepath.Ext(attachment.FileName), ".") - dataURL := nonEmpty(attachment.FileURL, attachment.ExternalURL) - payload := map[string]any{ - "id": attachment.ID, - "message_id": attachment.MessageID, - "file_type": attachment.FileType, - "account_id": attachment.AccountID, - "data_url": dataURL, - "thumb_url": attachment.ThumbURL, - "file_size": attachment.FileSize, - "extension": extension, - "width": attachment.Width, - "height": attachment.Height, - } - if strings.TrimSpace(attachment.Metadata) != "" { - metadata := map[string]any{} - if json.Unmarshal([]byte(attachment.Metadata), &metadata) == nil { - payload["metadata"] = metadata - } - } - return payload + return attachment.PushEventData() } func serializeAttachmentWithConversation(ctx context.Context, db *gorm.DB, attachment *model.Attachment) map[string]any { @@ -662,27 +641,27 @@ func serializeMessageSender(ctx context.Context, db *gorm.DB, message *model.Mes if senderType == "contact" { var contact model.Contact if err := db.WithContext(ctx).First(&contact, *message.SenderID).Error; err == nil { - return serializeContactWithContext(ctx, &contact) + return contact.PushEventData() } return nil } if senderType == "agent_bot" { var bot model.AgentBot if err := db.WithContext(ctx).First(&bot, *message.SenderID).Error; err == nil { - return serializeAgentBotSender(&bot) + return bot.PushEventData() } return nil } if senderType == "captain_assistant" { var assistant model.CaptainAssistant if err := db.WithContext(ctx).First(&assistant, *message.SenderID).Error; err == nil { - return serializeCaptainAssistantSender(&assistant) + return assistant.PushEventData() } return nil } var user model.User if err := db.WithContext(ctx).First(&user, *message.SenderID).Error; err == nil { - return serializeUser(&user, message.AccountID) + return user.PushEventData() } return nil } @@ -764,11 +743,6 @@ func serializeAgentBotSender(bot *model.AgentBot) map[string]any { } } -func serializeCaptainAssistantSender(assistant *model.CaptainAssistant) map[string]any { - avatarURL := strings.TrimRight(os.Getenv("FRONTEND_URL"), "/") + "/assets/images/dashboard/captain/logo.svg" - return map[string]any{"id": assistant.ID, "name": assistant.Name, "avatar_url": avatarURL, "description": assistant.Description, "created_at": assistant.CreatedAt, "type": "captain_assistant"} -} - func serializeAgentBotSlim(bot *model.AgentBot) map[string]any { return map[string]any{ "id": bot.ID, diff --git a/backend/internal/handler/api/v1/message_handler_test.go b/backend/internal/handler/api/v1/message_handler_test.go index b46c9e8a..afcd4ba1 100644 --- a/backend/internal/handler/api/v1/message_handler_test.go +++ b/backend/internal/handler/api/v1/message_handler_test.go @@ -495,7 +495,8 @@ func (s *MessageHandlerTestSuite) TestCreate_ChatwootFrontendPayloadDefaultsOutg assert.Equal(s.T(), float64(s.testUser.ID), sender["id"]) assert.Equal(s.T(), "Msg Handler Agent", sender["name"]) assert.Equal(s.T(), "Message Agent", sender["available_name"]) - assert.Equal(s.T(), "message-agent@example.com", sender["email"]) + assert.Equal(s.T(), "user", sender["type"]) + assert.NotContains(s.T(), sender, "email") } func (s *MessageHandlerTestSuite) TestCreate_MultipartAttachmentPersistsAndSerializes() { diff --git a/backend/internal/model/attachment.go b/backend/internal/model/attachment.go index 04842b39..63494e55 100644 --- a/backend/internal/model/attachment.go +++ b/backend/internal/model/attachment.go @@ -1,5 +1,11 @@ package model +import ( + "encoding/json" + "path/filepath" + "strings" +) + // Attachment represents a file attachment on a message. // Reference: Chatwoot Attachment model + P2B M3 spec type Attachment struct { @@ -21,4 +27,29 @@ type Attachment struct { Account Account `gorm:"foreignKey:AccountID" json:"account,omitempty"` } -func (Attachment) TableName() string { return "attachments" } \ No newline at end of file +func (Attachment) TableName() string { return "attachments" } + +// PushEventData matches Chatwoot's attachment payload shared by HTTP and WebSocket messages. +func (a *Attachment) PushEventData() map[string]any { + dataURL := a.FileURL + if dataURL == "" { + dataURL = a.ExternalURL + } + payload := map[string]any{ + "id": a.ID, + "message_id": a.MessageID, + "file_type": a.FileType, + "account_id": a.AccountID, + "data_url": dataURL, + "thumb_url": a.ThumbURL, + "file_size": a.FileSize, + "extension": strings.TrimPrefix(filepath.Ext(a.FileName), "."), + "width": a.Width, + "height": a.Height, + } + metadata := map[string]any{} + if json.Unmarshal([]byte(a.Metadata), &metadata) == nil && len(metadata) > 0 { + payload["metadata"] = metadata + } + return payload +} diff --git a/backend/internal/model/captain_models.go b/backend/internal/model/captain_models.go index c310cd90..598d3bf3 100644 --- a/backend/internal/model/captain_models.go +++ b/backend/internal/model/captain_models.go @@ -3,6 +3,8 @@ package model import ( "encoding/json" "fmt" + "os" + "strings" "github.com/pgvector/pgvector-go" ) @@ -89,6 +91,19 @@ type CaptainAssistant struct { func (CaptainAssistant) TableName() string { return "captain_assistants" } +// PushEventData matches Chatwoot's public message-sender contract. +func (a *CaptainAssistant) PushEventData() map[string]any { + avatarURL := strings.TrimRight(os.Getenv("FRONTEND_URL"), "/") + "/assets/images/dashboard/captain/logo.svg" + return map[string]any{ + "id": a.ID, + "name": a.Name, + "avatar_url": avatarURL, + "description": a.Description, + "created_at": a.CreatedAt, + "type": "captain_assistant", + } +} + // DefaultAssistantConfig returns the default configuration for a new Captain Assistant. // Reference: Chatwoot Captain::Assistant default config values func DefaultAssistantConfig() map[string]interface{} { diff --git a/backend/internal/model/contact.go b/backend/internal/model/contact.go index 15f863d8..77b13c92 100644 --- a/backend/internal/model/contact.go +++ b/backend/internal/model/contact.go @@ -34,6 +34,22 @@ type Contact struct { func (Contact) TableName() string { return "contacts" } +// PushEventData matches Chatwoot's public message-sender contract. +func (c *Contact) PushEventData() map[string]any { + return map[string]any{ + "additional_attributes": jsonMap(c.AdditionalAttributes), + "custom_attributes": jsonMap(c.CustomAttributes), + "email": c.Email, + "id": c.ID, + "identifier": c.Identifier, + "name": c.Name, + "phone_number": c.PhoneNumber, + "thumbnail": c.AvatarURL, + "blocked": c.Blocked, + "type": "contact", + } +} + // BeforeSave syncs location/country_code from additional_attributes and upgrades contact_type. // Reference: Chatwoot before_save :sync_contact_attributes → Contacts::SyncAttributes func (c *Contact) BeforeSave(tx *gorm.DB) error { diff --git a/backend/internal/model/user.go b/backend/internal/model/user.go index fad9406d..72d8b5f4 100644 --- a/backend/internal/model/user.go +++ b/backend/internal/model/user.go @@ -47,6 +47,27 @@ type User struct { func (User) TableName() string { return "users" } +// PushEventData matches Chatwoot's public message-sender contract. +func (u *User) PushEventData() map[string]any { + availableName := u.DisplayName + if availableName == "" { + availableName = u.Name + } + availabilityStatus := "offline" + if u.Available { + availabilityStatus = "online" + } + return map[string]any{ + "id": u.ID, + "name": u.Name, + "available_name": availableName, + "avatar_url": u.AvatarURL, + "type": "user", + "availability_status": availabilityStatus, + "thumbnail": u.AvatarURL, + } +} + // BeforeCreate mirrors Chatwoot's Pubsubable concern by assigning every user a cable token. func (u *User) BeforeCreate(tx *gorm.DB) error { if u.PubsubToken == "" { diff --git a/backend/internal/service/message_service.go b/backend/internal/service/message_service.go index 97d36155..b4b54015 100644 --- a/backend/internal/service/message_service.go +++ b/backend/internal/service/message_service.go @@ -422,9 +422,14 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint if req.IsVoiceMessage && fileType == "audio" { attachment.Metadata = `{"is_voice_message":true}` } - if err := tx.Create(attachment).Error; err != nil { + createAttachment := tx + if attachment.Metadata == "" { + createAttachment = tx.Omit("Metadata") + } + if err := createAttachment.Create(attachment).Error; err != nil { return err } + message.Attachments = append(message.Attachments, *attachment) } if queueShangwutong && s.worker != nil { var err error diff --git a/backend/internal/service/message_service_test.go b/backend/internal/service/message_service_test.go index caa901f5..c96303ec 100644 --- a/backend/internal/service/message_service_test.go +++ b/backend/internal/service/message_service_test.go @@ -375,6 +375,7 @@ func TestMessageService_Create(t *testing.T) { Content: "测试消息内容", MessageType: "outgoing", ContentType: "text", + Attachments: []MessageAttachmentInput{{FileName: "fanout.txt", FileSize: 17, ContentType: "text/plain"}}, } created, err := svc.Create(ctx, account.ID, user.ID, req) assert.NoError(t, err) @@ -386,6 +387,8 @@ func TestMessageService_Create(t *testing.T) { assert.Equal(t, user.ID, *created.SenderID) assert.Equal(t, "user", created.SenderType) assert.False(t, created.Private) + require.Len(t, created.Attachments, 1) + assert.Empty(t, created.Attachments[0].Metadata) var updatedConversation model.Conversation require.NoError(t, db.First(&updatedConversation, conv.ID).Error) require.NotNil(t, updatedConversation.LastActivityAt) diff --git a/backend/internal/wsevent/bridge_listener.go b/backend/internal/wsevent/bridge_listener.go index e664ffad..9cc86524 100644 --- a/backend/internal/wsevent/bridge_listener.go +++ b/backend/internal/wsevent/bridge_listener.go @@ -137,9 +137,23 @@ func messagePushPayload(message *model.Message, data map[string]interface{}) map payload["message_type"] = messageTypeValue(message.MessageType) payload["content_type"] = nonEmpty(message.ContentType, "text") payload["status"] = nonEmpty(message.Status, "sent") + payload["source_id"] = message.SourceID if message.ContentAttributes == nil { payload["content_attributes"] = map[string]interface{}{} } + if message.AdditionalAttributes == nil { + payload["additional_attributes"] = map[string]interface{}{} + } + if message.ExternalSourceIDs == nil { + payload["external_source_ids"] = map[string]interface{}{} + } + if len(message.Attachments) > 0 { + attachments := make([]any, 0, len(message.Attachments)) + for i := range message.Attachments { + attachments = append(attachments, message.Attachments[i].PushEventData()) + } + payload["attachments"] = attachments + } conversationID := message.ConversationID conversationPayload := map[string]interface{}{ @@ -160,20 +174,14 @@ func messagePushPayload(message *model.Message, data map[string]interface{}) map "source_id": contact.Identifier, } if senderType == "Contact" { - payload["sender"] = contactPushPayload(contact) + payload["sender"] = contact.PushEventData() } } payload["conversation_id"] = conversationID payload["conversation"] = conversationPayload - if senderType == "User" || senderType == "AgentBot" { + if senderType == "User" || senderType == "AgentBot" || senderType == "Captain::Assistant" { if sender, ok := eventSender(data); ok { - expectedType := "user" - if senderType == "AgentBot" { - expectedType = "agent_bot" - } - if sender["type"] == expectedType { - payload["sender"] = sender - } + payload["sender"] = sender } } payload["sender_type"] = senderType @@ -186,15 +194,7 @@ func eventSender(data map[string]interface{}) (map[string]interface{}, bool) { 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 + return sender.PushEventData(), true case model.User: return eventSender(map[string]interface{}{"sender": &sender}) case *model.AgentBot: @@ -208,9 +208,7 @@ func eventSender(data map[string]interface{}) (map[string]interface{}, bool) { if sender == nil { return nil, false } - return map[string]interface{}{ - "id": sender.ID, "name": sender.Name, "type": "agent_bot", - }, true + return sender.PushEventData(), true case model.CaptainAssistant: return eventSender(map[string]interface{}{"sender": &sender}) case map[string]interface{}: @@ -310,8 +308,10 @@ func senderTypeName(value string) string { switch strings.ToLower(trimmed) { case "contact": return "Contact" - case "agentbot", "agent_bot", "captain::assistant", "captainassistant", "captain_assistant": + case "agentbot", "agent_bot": return "AgentBot" + case "captain::assistant", "captainassistant", "captain_assistant": + return "Captain::Assistant" case "user": return "User" default: diff --git a/backend/internal/wsevent/bridge_listener_test.go b/backend/internal/wsevent/bridge_listener_test.go index e391d5dd..9ab6d10f 100644 --- a/backend/internal/wsevent/bridge_listener_test.go +++ b/backend/internal/wsevent/bridge_listener_test.go @@ -68,8 +68,8 @@ func TestBridgeListenerPreservesWebWidgetSenderTypeContract(t *testing.T) { 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: "Captain::Assistant", expectedType: "Captain::Assistant", sender: &model.CaptainAssistant{Base: model.Base{ID: 8}, Name: "Captain"}, expectedName: "Captain"}, + {senderType: "CaptainAssistant", expectedType: "Captain::Assistant", sender: &model.CaptainAssistant{Base: model.Base{ID: 8}, Name: "Captain"}, expectedName: "Captain"}, {senderType: "CustomSender", expectedType: "CustomSender"}, } for _, tt := range tests { @@ -105,7 +105,11 @@ func TestBridgeListenerPreservesWebWidgetSenderTypeContract(t *testing.T) { return } sender, ok := payload["sender"].(map[string]interface{}) - if !ok || sender["name"] != tt.expectedName || sender["type"] != "agent_bot" { + expectedSenderObjectType := "agent_bot" + if tt.expectedType == "Captain::Assistant" { + expectedSenderObjectType = "captain_assistant" + } + if !ok || sender["name"] != tt.expectedName || sender["type"] != expectedSenderObjectType { t.Fatalf("sender does not match sender_type: %#v", payload) } }) diff --git a/backend/tests/e2e/websocket_multi_instance_e2e_test.go b/backend/tests/e2e/websocket_multi_instance_e2e_test.go index bacf05d5..626cff47 100644 --- a/backend/tests/e2e/websocket_multi_instance_e2e_test.go +++ b/backend/tests/e2e/websocket_multi_instance_e2e_test.go @@ -6,6 +6,7 @@ import ( "encoding/json" "fmt" "io" + "mime/multipart" "net" "net/http" "net/url" @@ -23,9 +24,12 @@ import ( type fanoutSeed struct { AccountID uint `json:"account_id"` + ConversationID uint `json:"conversation_id"` ConversationDisplayID uint `json:"conversation_display_id"` + CaptainAssistantID uint `json:"captain_assistant_id"` AdminEmail string `json:"admin_email"` AdminPassword string `json:"admin_password"` + WidgetToken string `json:"widget_token"` } type testProcess struct { @@ -49,9 +53,12 @@ func TestWebSocketMultiInstanceFanout(t *testing.T) { require.NotEmpty(t, redisDSN, "GOCHAT_REDIS_DSN is required") root := moduleRoot(t) + stamp := strconv.FormatInt(time.Now().UnixNano(), 10) evidenceDir := os.Getenv("GOCHAT_WS_E2E_EVIDENCE_DIR") if evidenceDir == "" { evidenceDir = t.TempDir() + } else { + evidenceDir = filepath.Join(evidenceDir, stamp) } require.NoError(t, os.MkdirAll(evidenceDir, 0o755)) @@ -73,7 +80,6 @@ func TestWebSocketMultiInstanceFanout(t *testing.T) { t.Fatalf("build GoChat: %v\n%s", err, output) } - stamp := strconv.FormatInt(time.Now().UnixNano(), 10) baseEnv := map[string]string{ "GOCHAT_ENV": "development", "GOCHAT_DATABASE_DSN": databaseDSN, @@ -102,18 +108,10 @@ func TestWebSocketMultiInstanceFanout(t *testing.T) { serverEnv := cloneMap(baseEnv) serverEnv["GOCHAT_DATABASE_RUN_MIGRATIONS"] = "false" serverEnv["GOCHAT_SERVER_HOST"] = "127.0.0.1" - portA, portB := freePort(t), freePort(t) - for portB == portA { - portB = freePort(t) - } - baseURLA := fmt.Sprintf("http://127.0.0.1:%d", portA) - baseURLB := fmt.Sprintf("http://127.0.0.1:%d", portB) - instanceA := startTestProcess(t, root, binary, portA, serverEnv, filepath.Join(evidenceDir, "instance-a.log")) - instanceB := startTestProcess(t, root, binary, portB, serverEnv, filepath.Join(evidenceDir, "instance-b.log")) - t.Cleanup(instanceB.stop) + instanceA, baseURLA := startHealthyTestProcess(t, root, binary, serverEnv, filepath.Join(evidenceDir, "instance-a.log")) t.Cleanup(instanceA.stop) - waitForHealth(t, baseURLA) - waitForHealth(t, baseURLB) + instanceB, baseURLB := startHealthyTestProcess(t, root, binary, serverEnv, filepath.Join(evidenceDir, "instance-b.log")) + t.Cleanup(instanceB.stop) authHeaders := signIn(t, baseURLA, seedData.AdminEmail, seedData.AdminPassword) connA, identifier := connectAccountCable(t, baseURLA, issueWSTicket(t, baseURLA, authHeaders), seedData.AccountID) @@ -122,18 +120,46 @@ func TestWebSocketMultiInstanceFanout(t *testing.T) { defer connB.Close() require.JSONEq(t, identifier, identifierB) - content := "multi-instance fanout " + stamp - created := createMessage(t, baseURLA, authHeaders, seedData, content) - frameA := readMessageCreated(t, connA, content) - frameB := readMessageCreated(t, connB, content) + agentBotID := startAITakeover(t, baseURLA, authHeaders, seedData) + senderCases := []struct { + name string + senderType string + senderObjectType string + senderID uint + widget bool + attachment bool + }{ + {name: "agent_bot", senderType: "AgentBot", senderObjectType: "agent_bot", senderID: agentBotID}, + {name: "contact", senderType: "Contact", senderObjectType: "contact", widget: true}, + {name: "captain_assistant", senderType: "Captain::Assistant", senderObjectType: "captain_assistant", senderID: seedData.CaptainAssistantID}, + {name: "user", senderType: "User", senderObjectType: "user", attachment: true}, + } + frames := make(map[string]any, len(senderCases)) + for _, senderCase := range senderCases { + content := "multi-instance " + senderCase.name + " " + stamp + var created map[string]any + if senderCase.widget { + createWidgetMessage(t, baseURLA, seedData, content) + } else { + created = createDashboardMessage(t, baseURLA, authHeaders, seedData, content, senderCase.senderType, senderCase.senderID, senderCase.attachment) + } + frameA := readMessageCreated(t, connA, content) + frameB := readMessageCreated(t, connB, content) + if created == nil { + created = fetchHTTPMessage(t, baseURLA, authHeaders, seedData, content) + } - require.Equal(t, frameA, frameB, "both processes must deliver the complete identical frame") - assertFanoutContract(t, frameA, identifier, created, seedData, content) + require.Equal(t, frameA, frameB, "%s sender must fan out identically", senderCase.name) + assertFanoutContract(t, frameA, identifier, created, seedData, content, senderCase.senderType, senderCase.senderObjectType, !senderCase.widget, senderCase.attachment) + assertFanoutContract(t, frameB, identifier, created, seedData, content, senderCase.senderType, senderCase.senderObjectType, !senderCase.widget, senderCase.attachment) + frames[senderCase.name+"_a"] = frameA + frames[senderCase.name+"_b"] = frameB + } report["status"] = "passed" report["instance_a"] = baseURLA report["instance_b"] = baseURLB - report["frame"] = frameA + report["frames"] = frames } func moduleRoot(t *testing.T) string { @@ -190,7 +216,9 @@ func freePort(t *testing.T) int { func startTestProcess(t *testing.T, root, binary string, port int, baseEnv map[string]string, logPath string) *testProcess { t.Helper() - logFile, err := os.Create(logPath) + logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) + require.NoError(t, err) + _, err = fmt.Fprintf(logFile, "\n=== start attempt on port %d ===\n", port) require.NoError(t, err) ctx, cancel := context.WithCancel(context.Background()) cmd := exec.CommandContext(ctx, binary, "serve") @@ -205,6 +233,21 @@ func startTestProcess(t *testing.T, root, binary string, port int, baseEnv map[s return &testProcess{cancel: cancel, cmd: cmd, done: done, log: logFile} } +func startHealthyTestProcess(t *testing.T, root, binary string, baseEnv map[string]string, logPath string) (*testProcess, string) { + t.Helper() + for attempt := 1; attempt <= 5; attempt++ { + port := freePort(t) + baseURL := fmt.Sprintf("http://127.0.0.1:%d", port) + process := startTestProcess(t, root, binary, port, baseEnv, logPath) + if waitForHealth(baseURL, process, 45*time.Second) { + return process, baseURL + } + process.stop() + } + t.Fatalf("GoChat failed to claim a free port after 5 attempts; see %s", logPath) + return nil, "" +} + func (process *testProcess) stop() { process.cancel() select { @@ -216,21 +259,26 @@ func (process *testProcess) stop() { _ = process.log.Close() } -func waitForHealth(t *testing.T, baseURL string) { - t.Helper() +func waitForHealth(baseURL string, process *testProcess, timeout time.Duration) bool { client := &http.Client{Timeout: time.Second} - deadline := time.Now().Add(45 * time.Second) + deadline := time.Now().Add(timeout) for time.Now().Before(deadline) { response, err := client.Get(baseURL + "/health") if err == nil { response.Body.Close() if response.StatusCode == http.StatusOK { - return + return true } } + select { + case processErr := <-process.done: + process.done <- processErr + return false + default: + } time.Sleep(250 * time.Millisecond) } - t.Fatalf("GoChat did not become healthy: %s", baseURL) + return false } func signIn(t *testing.T, baseURL, email, password string) http.Header { @@ -308,26 +356,114 @@ func responseStatus(response *http.Response) any { return response.StatusCode } -func createMessage(t *testing.T, baseURL string, authHeaders http.Header, seed fanoutSeed, content string) map[string]any { +func startAITakeover(t *testing.T, baseURL string, authHeaders http.Header, seed fanoutSeed) uint { t.Helper() - body, err := json.Marshal(map[string]any{"content": content, "message_type": "outgoing", "private": false}) + endpoint := fmt.Sprintf("%s/api/v1/accounts/%d/conversations/%d/ai_takeover", baseURL, seed.AccountID, seed.ConversationDisplayID) + request, err := http.NewRequest(http.MethodPost, endpoint, nil) require.NoError(t, err) + setAuthHeaders(request, authHeaders) + response, err := http.DefaultClient.Do(request) + require.NoError(t, err) + defer response.Body.Close() + responseBody, err := io.ReadAll(response.Body) + require.NoError(t, err) + require.Equal(t, http.StatusOK, response.StatusCode, "start AI takeover response: %s", responseBody) + var conversation map[string]any + require.NoError(t, json.Unmarshal(responseBody, &conversation)) + assignee := conversation["meta"].(map[string]any)["assignee"].(map[string]any) + return uint(assignee["id"].(float64)) +} + +func createDashboardMessage(t *testing.T, baseURL string, authHeaders http.Header, seed fanoutSeed, content, senderType string, senderID uint, attachment bool) map[string]any { + t.Helper() + var body bytes.Buffer + writer := multipart.NewWriter(&body) + for key, value := range map[string]string{ + "content": content, + "message_type": "outgoing", + "private": "false", + "source_id": "ws-e2e-source", + "echo_id": "ws-e2e-echo", + "content_attributes": `{"e2e":{"enabled":true,"levels":[1,"two",{"deep":null}]}}`, + "additional_attributes": `{"e2e":{"score":2.5,"tags":["fanout",7]}}`, + "external_source_ids": `{"e2e":{"source":"fanout","ids":["a",2]}}`, + } { + require.NoError(t, writer.WriteField(key, value)) + } + if senderType != "User" { + require.NoError(t, writer.WriteField("sender_type", senderType)) + require.NoError(t, writer.WriteField("sender_id", strconv.FormatUint(uint64(senderID), 10))) + } + if attachment { + part, err := writer.CreateFormFile("attachments[]", "fanout.txt") + require.NoError(t, err) + _, err = part.Write([]byte("fanout attachment")) + require.NoError(t, err) + } + require.NoError(t, writer.Close()) endpoint := fmt.Sprintf("%s/api/v1/accounts/%d/conversations/%d/messages", baseURL, seed.AccountID, seed.ConversationDisplayID) - request, err := http.NewRequest(http.MethodPost, endpoint, bytes.NewReader(body)) + request, err := http.NewRequest(http.MethodPost, endpoint, &body) + require.NoError(t, err) + request.Header.Set("Content-Type", writer.FormDataContentType()) + setAuthHeaders(request, authHeaders) + response, err := http.DefaultClient.Do(request) + require.NoError(t, err) + defer response.Body.Close() + responseBody, err := io.ReadAll(response.Body) + require.NoError(t, err) + require.Equal(t, http.StatusOK, response.StatusCode, "create message response: %s", responseBody) + var created map[string]any + require.NoError(t, json.Unmarshal(responseBody, &created)) + return created +} + +func createWidgetMessage(t *testing.T, baseURL string, seed fanoutSeed, content string) { + t.Helper() + body, err := json.Marshal(map[string]any{"content": content, "conversation_id": seed.ConversationID}) + require.NoError(t, err) + request, err := http.NewRequest(http.MethodPost, baseURL+"/widget/messages", bytes.NewReader(body)) require.NoError(t, err) request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-Widget-Token", seed.WidgetToken) + response, err := http.DefaultClient.Do(request) + require.NoError(t, err) + defer response.Body.Close() + responseBody, err := io.ReadAll(response.Body) + require.NoError(t, err) + require.Equal(t, http.StatusOK, response.StatusCode, "create widget message response: %s", responseBody) +} + +func fetchHTTPMessage(t *testing.T, baseURL string, authHeaders http.Header, seed fanoutSeed, content string) map[string]any { + t.Helper() + endpoint := fmt.Sprintf("%s/api/v1/accounts/%d/conversations/%d/messages", baseURL, seed.AccountID, seed.ConversationDisplayID) + request, err := http.NewRequest(http.MethodGet, endpoint, nil) + require.NoError(t, err) + setAuthHeaders(request, authHeaders) + response, err := http.DefaultClient.Do(request) + require.NoError(t, err) + defer response.Body.Close() + responseBody, err := io.ReadAll(response.Body) + require.NoError(t, err) + require.Equal(t, http.StatusOK, response.StatusCode, "list messages response: %s", responseBody) + var result struct { + Payload []map[string]any `json:"payload"` + } + require.NoError(t, json.Unmarshal(responseBody, &result)) + for _, message := range result.Payload { + if message["content"] == content { + return message + } + } + t.Fatalf("HTTP message response did not contain %q", content) + return nil +} + +func setAuthHeaders(request *http.Request, authHeaders http.Header) { for _, name := range []string{"access-token", "client", "uid", "token-type"} { if value := authHeaders.Get(name); value != "" { request.Header.Set(name, value) } } - response, err := http.DefaultClient.Do(request) - require.NoError(t, err) - defer response.Body.Close() - require.Equal(t, http.StatusOK, response.StatusCode) - var created map[string]any - require.NoError(t, json.NewDecoder(response.Body).Decode(&created)) - return created } func readMessageCreated(t *testing.T, conn *websocket.Conn, content string) map[string]any { @@ -346,7 +482,7 @@ func readMessageCreated(t *testing.T, conn *websocket.Conn, content string) map[ } } -func assertFanoutContract(t *testing.T, frame map[string]any, identifier string, created map[string]any, seed fanoutSeed, content string) { +func assertFanoutContract(t *testing.T, frame map[string]any, identifier string, created map[string]any, seed fanoutSeed, content, senderType, senderObjectType string, structured, attachment bool) { t.Helper() require.ElementsMatch(t, []string{"identifier", "message"}, mapKeys(frame)) require.JSONEq(t, identifier, frame["identifier"].(string)) @@ -355,18 +491,72 @@ func assertFanoutContract(t *testing.T, frame map[string]any, identifier string, require.Equal(t, "message.created", message["event"]) require.Equal(t, float64(seed.AccountID), message["account_id"]) payload := message["data"].(map[string]any) - for _, key := range []string{ - "id", "account_id", "inbox_id", "conversation_id", "content", "message_type", - "content_type", "status", "private", "external", "sender_type", "created_at", "conversation", - } { - require.Contains(t, payload, key) - } - for _, key := range []string{"id", "account_id", "inbox_id", "conversation_id", "content", "message_type", "content_type", "status", "private", "external"} { - require.Equal(t, created[key], payload[key], "payload field %s must match the HTTP contract", key) - } + assertWSTransportFields(t, payload, created, senderType) + require.Equal(t, normalizedHTTPMessage(t, created), normalizedWSMessage(t, payload), "normalized HTTP and WebSocket message contracts must match completely") require.Equal(t, content, payload["content"]) - require.Equal(t, "User", payload["sender_type"]) - require.Contains(t, payload["conversation"].(map[string]any), "last_activity_at") + assertNestedFrontendFields(t, payload, senderObjectType, structured, attachment) +} + +func assertWSTransportFields(t *testing.T, payload, created map[string]any, senderType string) { + t.Helper() + require.Equal(t, senderType, payload["sender_type"]) + httpSender := created["sender"].(map[string]any) + require.Equal(t, httpSender["id"], payload["sender_id"]) + require.Nil(t, payload["deleted_at"]) + _, err := time.Parse(time.RFC3339Nano, payload["updated_at"].(string)) + require.NoError(t, err) + conversation := payload["conversation"].(map[string]any) + require.ElementsMatch(t, []string{"assignee_id", "contact_inbox", "last_activity_at"}, mapKeys(conversation)) + require.IsType(t, float64(0), conversation["last_activity_at"]) + contactInbox := conversation["contact_inbox"].(map[string]any) + require.IsType(t, "", contactInbox["source_id"]) +} + +func normalizedHTTPMessage(t *testing.T, message map[string]any) map[string]any { + t.Helper() + return cloneJSONMap(t, message) +} + +func normalizedWSMessage(t *testing.T, message map[string]any) map[string]any { + t.Helper() + normalized := cloneJSONMap(t, message) + for _, key := range []string{"conversation", "deleted_at", "sender_id", "sender_type", "updated_at"} { + delete(normalized, key) + } + return normalized +} + +func cloneJSONMap(t *testing.T, value map[string]any) map[string]any { + t.Helper() + raw, err := json.Marshal(value) + require.NoError(t, err) + var clone map[string]any + require.NoError(t, json.Unmarshal(raw, &clone)) + return clone +} + +func assertNestedFrontendFields(t *testing.T, payload map[string]any, senderObjectType string, structured, attachment bool) { + t.Helper() + sender := payload["sender"].(map[string]any) + require.NotEmpty(t, sender["name"].(string)) + require.Equal(t, senderObjectType, sender["type"]) + if structured { + contentAttributes := payload["content_attributes"].(map[string]any) + require.IsType(t, []any{}, contentAttributes["e2e"].(map[string]any)["levels"]) + additionalAttributes := payload["additional_attributes"].(map[string]any) + require.IsType(t, []any{}, additionalAttributes["e2e"].(map[string]any)["tags"]) + externalSourceIDs := payload["external_source_ids"].(map[string]any) + require.IsType(t, []any{}, externalSourceIDs["e2e"].(map[string]any)["ids"]) + } + if attachment { + attachments := payload["attachments"].([]any) + require.Len(t, attachments, 1) + item := attachments[0].(map[string]any) + require.Equal(t, "file", item["file_type"]) + require.Equal(t, "txt", item["extension"]) + require.IsType(t, float64(0), item["file_size"]) + } + require.NotContains(t, payload, "call", "non-voice messages must not expose a call object") } func mapKeys(value map[string]any) []string {