Rogeeandrogee
60a6ac4785
* fix: restore web widget realtime replies (HH-556) * fix: make realtime delivery durable (HH-556) * fix: make realtime message outbox atomic (HH-556) --------- Co-authored-by: Rogee <rogee@ipao.vip>
1383 lines
56 KiB
Go
1383 lines
56 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
"github.com/stretchr/testify/assert"
|
||
"github.com/stretchr/testify/require"
|
||
"gorm.io/datatypes"
|
||
"gorm.io/gorm"
|
||
|
||
"github.com/gochat/gochat/internal/channel"
|
||
"github.com/gochat/gochat/internal/llm"
|
||
"github.com/gochat/gochat/internal/model"
|
||
"github.com/gochat/gochat/internal/repository"
|
||
"github.com/gochat/gochat/internal/search"
|
||
"github.com/gochat/gochat/internal/worker"
|
||
wspkg "github.com/gochat/gochat/internal/ws"
|
||
"github.com/gochat/gochat/internal/wsevent"
|
||
)
|
||
|
||
// mockMessageLLMProvider implements llm.Provider for message service testing.
|
||
// (Separate name to avoid conflict with mockLLMProvider in captain_task_service_test.go)
|
||
type mockMessageLLMProvider struct {
|
||
chatResponse *llm.ChatResponse
|
||
chatErr error
|
||
chatCalls int
|
||
}
|
||
|
||
func (m *mockMessageLLMProvider) ChatCompletion(ctx context.Context, req llm.ChatRequest) (*llm.ChatResponse, error) {
|
||
m.chatCalls++
|
||
if m.chatErr != nil {
|
||
return nil, m.chatErr
|
||
}
|
||
return m.chatResponse, nil
|
||
}
|
||
|
||
func (m *mockMessageLLMProvider) CreateEmbedding(ctx context.Context, req llm.EmbeddingRequest) (*llm.EmbeddingResponse, error) {
|
||
return nil, fmt.Errorf("not implemented")
|
||
}
|
||
|
||
func (m *mockMessageLLMProvider) ChatCompletionStream(ctx context.Context, req llm.ChatRequest, onChunk func(llm.StreamChunk) error) error {
|
||
return fmt.Errorf("not implemented")
|
||
}
|
||
|
||
// mockRetryListener implements channel.EventListener for capturing dispatched events.
|
||
type mockRetryListener struct {
|
||
received bool
|
||
lastEventType channel.EventType
|
||
lastAccountID uint
|
||
lastInboxID uint
|
||
lastUserID uint
|
||
lastConvID uint
|
||
lastData map[string]interface{}
|
||
}
|
||
|
||
type messageEventListener struct {
|
||
events []*channel.ChannelEvent
|
||
err error
|
||
}
|
||
|
||
func (l *messageEventListener) Name() string { return "message-event-listener" }
|
||
|
||
func (l *messageEventListener) OnEvent(_ context.Context, event *channel.ChannelEvent) error {
|
||
l.events = append(l.events, event)
|
||
return l.err
|
||
}
|
||
|
||
func (l *mockRetryListener) Name() string {
|
||
return "mock-retry-listener"
|
||
}
|
||
|
||
func (l *mockRetryListener) OnEvent(ctx context.Context, event *channel.ChannelEvent) error {
|
||
l.received = true
|
||
l.lastEventType = event.Type
|
||
l.lastAccountID = event.AccountID
|
||
l.lastInboxID = event.InboxID
|
||
l.lastUserID = event.UserID
|
||
l.lastConvID = event.ConversationID
|
||
l.lastData = event.Data
|
||
return nil
|
||
}
|
||
|
||
// setupMessageService creates MessageRepo + Dispatcher + mockLLM + MessageService test instances.
|
||
func setupMessageService(t *testing.T, mockLLM *mockMessageLLMProvider) (*gorm.DB, *repository.MessageRepo, *channel.Dispatcher, *MessageService) {
|
||
t.Helper()
|
||
db := setupServiceTestDB(t)
|
||
repo := repository.NewMessageRepo(db)
|
||
dispatcher := channel.NewDispatcher()
|
||
svc := NewMessageService(repo, dispatcher, mockLLM)
|
||
return db, repo, dispatcher, svc
|
||
}
|
||
|
||
// setupMessageServiceWithDefaultLLM creates a MessageService with a default mock LLM that returns "translated text".
|
||
func setupMessageServiceWithDefaultLLM(t *testing.T) (*gorm.DB, *repository.MessageRepo, *channel.Dispatcher, *MessageService) {
|
||
t.Helper()
|
||
mockLLM := &mockMessageLLMProvider{
|
||
chatResponse: &llm.ChatResponse{
|
||
Choices: []llm.ChatChoice{
|
||
{Message: llm.ChatMessage{Role: "assistant", Content: "translated text"}},
|
||
},
|
||
},
|
||
}
|
||
return setupMessageService(t, mockLLM)
|
||
}
|
||
|
||
func TestMessageService_WebWidgetReplyCarriesRealtimeContext(t *testing.T) {
|
||
db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t)
|
||
pool := worker.NewWorkerPool(db, nil)
|
||
svc.SetWorkerPool(pool)
|
||
publisher := wspkg.NewEventPublisherLocal(nil, nil)
|
||
publisher.SetWorkerPool(pool)
|
||
bridge := wsevent.New(publisher)
|
||
dispatcher.Register(bridge)
|
||
svc.SetRealtimeEventBridge(bridge)
|
||
account := createTestAccount(t, db)
|
||
agent := createTestUser(t, db, account.ID)
|
||
inbox := createTestInbox(t, db, account.ID, string(channel.ChannelWebWidget))
|
||
contact := createTestContact(t, db, account.ID)
|
||
contactInbox := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, SourceID: "visitor", PubsubToken: "visitor-token"}
|
||
require.NoError(t, db.Create(contactInbox).Error)
|
||
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
require.NoError(t, db.Model(conversation).Update("contact_inbox_id", contactInbox.ID).Error)
|
||
|
||
listener := &messageEventListener{}
|
||
dispatcher.Register(listener)
|
||
_, err := svc.Create(context.Background(), account.ID, agent.ID, CreateMessageRequest{
|
||
ConversationID: conversation.ID,
|
||
Content: "agent reply",
|
||
ContentType: "text",
|
||
MessageType: "outgoing",
|
||
})
|
||
require.NoError(t, err)
|
||
|
||
var created *channel.ChannelEvent
|
||
for _, event := range listener.events {
|
||
if event.Type == channel.EventMessageCreated {
|
||
created = event
|
||
break
|
||
}
|
||
}
|
||
require.NotNil(t, created)
|
||
assert.Equal(t, channel.ChannelWebWidget, created.Channel)
|
||
assert.Equal(t, "visitor-token", created.Data["widget_token"])
|
||
assert.IsType(t, &model.Conversation{}, created.Data["conversation"])
|
||
assert.IsType(t, &model.Contact{}, created.Data["contact"])
|
||
assert.IsType(t, &model.User{}, created.Data["sender"])
|
||
var realtimeJobs int64
|
||
require.NoError(t, db.Model(&model.BackgroundJob{}).Where("idempotency_key LIKE ?", "realtime:message.created:%").Count(&realtimeJobs).Error)
|
||
require.Equal(t, int64(2), realtimeJobs)
|
||
}
|
||
|
||
func TestMessageService_WebWidgetReplyRequiresRealtimeContext(t *testing.T) {
|
||
db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t)
|
||
account := createTestAccount(t, db)
|
||
agent := createTestUser(t, db, account.ID)
|
||
inbox := createTestInbox(t, db, account.ID, string(channel.ChannelWebWidget))
|
||
contact := createTestContact(t, db, account.ID)
|
||
conversation := &model.Conversation{
|
||
AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID,
|
||
Status: string(model.ConversationStatusOpen), ChannelType: "web_widget", Channel: "web_widget",
|
||
}
|
||
require.NoError(t, db.Create(conversation).Error)
|
||
|
||
listener := &messageEventListener{}
|
||
dispatcher.Register(listener)
|
||
_, err := svc.Create(context.Background(), account.ID, agent.ID, CreateMessageRequest{
|
||
ConversationID: conversation.ID,
|
||
Content: "agent reply",
|
||
ContentType: "text",
|
||
MessageType: "outgoing",
|
||
})
|
||
require.ErrorContains(t, err, "contact inbox")
|
||
assert.Empty(t, listener.events)
|
||
}
|
||
|
||
func TestMessageService_ReturnsMessageDispatchError(t *testing.T) {
|
||
db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t)
|
||
account := createTestAccount(t, db)
|
||
agent := createTestUser(t, db, account.ID)
|
||
inbox := createTestInbox(t, db, account.ID, string(channel.ChannelAPI))
|
||
contact := createTestContact(t, db, account.ID)
|
||
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
dispatchErr := errors.New("realtime unavailable")
|
||
dispatcher.Register(&messageEventListener{err: dispatchErr})
|
||
_, err := svc.Create(context.Background(), account.ID, agent.ID, CreateMessageRequest{
|
||
ConversationID: conversation.ID,
|
||
Content: "agent reply",
|
||
ContentType: "text",
|
||
MessageType: "outgoing",
|
||
})
|
||
require.ErrorIs(t, err, dispatchErr)
|
||
}
|
||
|
||
func TestMessageService_RealtimeFailureDoesNotBlockSendReply(t *testing.T) {
|
||
db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t)
|
||
pool := worker.NewWorkerPoolWithOptions(db, worker.WithBackoff(func(int) time.Duration { return 0 }))
|
||
svc.SetWorkerPool(pool)
|
||
rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:1", MaxRetries: -1, DialTimeout: 10 * time.Millisecond})
|
||
t.Cleanup(func() { require.NoError(t, rdb.Close()) })
|
||
publisher := wspkg.NewEventPublisher(nil, nil, wspkg.NewBroadcastRelay(rdb, nil))
|
||
publisher.SetWorkerPool(pool)
|
||
bridge := wsevent.New(publisher)
|
||
dispatcher.Register(bridge)
|
||
svc.SetRealtimeEventBridge(bridge)
|
||
|
||
account := createTestAccount(t, db)
|
||
agent := createTestUser(t, db, account.ID)
|
||
inbox := createTestInbox(t, db, account.ID, string(channel.ChannelAPI))
|
||
contact := createTestContact(t, db, account.ID)
|
||
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
message, err := svc.Create(context.Background(), account.ID, agent.ID, CreateMessageRequest{
|
||
ConversationID: conversation.ID,
|
||
Content: "agent reply",
|
||
ContentType: "text",
|
||
MessageType: "outgoing",
|
||
})
|
||
require.NoError(t, err)
|
||
require.NotNil(t, message)
|
||
|
||
processed, err := pool.ProcessOne(context.Background())
|
||
require.True(t, processed)
|
||
require.ErrorContains(t, err, "connect: connection refused")
|
||
|
||
var persistedMessages int64
|
||
require.NoError(t, db.Model(&model.Message{}).Where("id = ?", message.ID).Count(&persistedMessages).Error)
|
||
require.Equal(t, int64(1), persistedMessages)
|
||
var realtimeJob, sendReplyJob model.BackgroundJob
|
||
require.NoError(t, db.Where("queue = ? AND idempotency_key LIKE ?", "events", "realtime:%").First(&realtimeJob).Error)
|
||
require.Equal(t, model.BackgroundJobStatusRetrying, realtimeJob.Status)
|
||
require.NotEmpty(t, realtimeJob.LastError)
|
||
require.NoError(t, db.Where("job_type = ?", TaskTypeMessageSendReply).First(&sendReplyJob).Error)
|
||
require.Equal(t, model.BackgroundJobStatusQueued, sendReplyJob.Status)
|
||
}
|
||
|
||
func TestMessageService_RealtimeJobFailureRollsBackMessage(t *testing.T) {
|
||
db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t)
|
||
pool := worker.NewWorkerPool(db, nil)
|
||
svc.SetWorkerPool(pool)
|
||
publisher := wspkg.NewEventPublisherLocal(nil, nil)
|
||
publisher.SetWorkerPool(pool)
|
||
bridge := wsevent.New(publisher)
|
||
dispatcher.Register(bridge)
|
||
svc.SetRealtimeEventBridge(bridge)
|
||
|
||
account := createTestAccount(t, db)
|
||
agent := createTestUser(t, db, account.ID)
|
||
inbox := createTestInbox(t, db, account.ID, string(channel.ChannelAPI))
|
||
contact := createTestContact(t, db, account.ID)
|
||
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
failBackgroundJobCreate(t, db)
|
||
|
||
_, err := svc.Create(context.Background(), account.ID, agent.ID, CreateMessageRequest{
|
||
ConversationID: conversation.ID, Content: "must roll back", ContentType: "text", MessageType: "outgoing",
|
||
})
|
||
require.ErrorContains(t, err, "forced background job insert failure")
|
||
|
||
var messages, jobs int64
|
||
require.NoError(t, db.Model(&model.Message{}).Where("content = ?", "must roll back").Count(&messages).Error)
|
||
require.NoError(t, db.Model(&model.BackgroundJob{}).Count(&jobs).Error)
|
||
require.Zero(t, messages)
|
||
require.Zero(t, jobs)
|
||
}
|
||
|
||
func TestMessageService_APIReplyCarriesRealtimeContext(t *testing.T) {
|
||
db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t)
|
||
account := createTestAccount(t, db)
|
||
agent := createTestUser(t, db, account.ID)
|
||
inbox := createTestInbox(t, db, account.ID, string(channel.ChannelAPI))
|
||
contact := createTestContact(t, db, account.ID)
|
||
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
listener := &messageEventListener{}
|
||
dispatcher.Register(listener)
|
||
_, err := svc.Create(context.Background(), account.ID, agent.ID, CreateMessageRequest{
|
||
ConversationID: conversation.ID,
|
||
Content: "agent reply",
|
||
ContentType: "text",
|
||
MessageType: "outgoing",
|
||
})
|
||
require.NoError(t, err)
|
||
|
||
var created *channel.ChannelEvent
|
||
for _, event := range listener.events {
|
||
if event.Type == channel.EventMessageCreated {
|
||
created = event
|
||
break
|
||
}
|
||
}
|
||
require.NotNil(t, created)
|
||
assert.Equal(t, channel.ChannelAPI, created.Channel)
|
||
assert.NotContains(t, created.Data, "widget_token")
|
||
assert.IsType(t, &model.Inbox{}, created.Data["inbox"])
|
||
assert.IsType(t, &model.Conversation{}, created.Data["conversation"])
|
||
assert.IsType(t, &model.Contact{}, created.Data["contact"])
|
||
assert.IsType(t, &model.User{}, created.Data["sender"])
|
||
}
|
||
|
||
func TestMessageService_MessageEventContextResolvesAutomatedSenders(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
require.NoError(t, db.AutoMigrate(&model.AgentBot{}, &model.CaptainAssistant{}))
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, string(channel.ChannelWebWidget))
|
||
contact := createTestContact(t, db, account.ID)
|
||
contactInbox := &model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, SourceID: "visitor", PubsubToken: "visitor-token"}
|
||
require.NoError(t, db.Create(contactInbox).Error)
|
||
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
require.NoError(t, db.Model(conversation).Update("contact_inbox_id", contactInbox.ID).Error)
|
||
|
||
bot := &model.AgentBot{AccountID: &account.ID, Name: "Reply Bot", BotType: "webhook"}
|
||
require.NoError(t, db.Create(bot).Error)
|
||
assistant := &model.CaptainAssistant{AccountID: account.ID, Name: "Captain", Config: json.RawMessage(`{}`), Status: model.AssistantStatusActive}
|
||
require.NoError(t, db.Create(assistant).Error)
|
||
|
||
tests := []struct {
|
||
name string
|
||
senderType string
|
||
senderID uint
|
||
expected interface{}
|
||
}{
|
||
{name: "agent bot", senderType: "AgentBot", senderID: bot.ID, expected: &model.AgentBot{}},
|
||
{name: "captain namespaced", senderType: "Captain::Assistant", senderID: assistant.ID, expected: &model.CaptainAssistant{}},
|
||
{name: "captain alias", senderType: "CaptainAssistant", senderID: assistant.ID, expected: &model.CaptainAssistant{}},
|
||
{name: "unknown", senderType: "CustomSender", senderID: assistant.ID},
|
||
}
|
||
for _, tt := range tests {
|
||
t.Run(tt.name, func(t *testing.T) {
|
||
for _, channelType := range []channel.ChannelType{channel.ChannelWebWidget, channel.ChannelAPI} {
|
||
t.Run(string(channelType), func(t *testing.T) {
|
||
require.NoError(t, db.Model(inbox).Update("channel_type", string(channelType)).Error)
|
||
event := channel.NewChannelEvent(channel.EventMessageCreated, channelType, account.ID, inbox.ID)
|
||
message := &model.Message{AccountID: account.ID, ConversationID: conversation.ID, InboxID: inbox.ID, SenderID: &tt.senderID, SenderType: tt.senderType}
|
||
require.NoError(t, svc.addMessageEventContext(context.Background(), db, event, message))
|
||
|
||
assert.IsType(t, &model.Contact{}, event.Data["contact"])
|
||
if channelType == channel.ChannelWebWidget {
|
||
assert.Equal(t, "visitor-token", event.Data["widget_token"])
|
||
} else {
|
||
assert.NotContains(t, event.Data, "widget_token")
|
||
}
|
||
if tt.expected == nil {
|
||
assert.NotContains(t, event.Data, "sender")
|
||
} else {
|
||
assert.IsType(t, tt.expected, event.Data["sender"])
|
||
}
|
||
})
|
||
}
|
||
})
|
||
}
|
||
}
|
||
|
||
// ========== ListByConversation 测试 ==========
|
||
|
||
func TestMessageService_ListByConversation(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
ctx := context.Background()
|
||
account := createTestAccount(t, db)
|
||
user := createTestUser(t, db, account.ID)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
// 创建多条消息
|
||
msg1 := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "消息1", MessageType: "incoming", ContentType: "text", SenderType: "contact", SenderID: &contact.ID,
|
||
}
|
||
msg2 := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "消息2", MessageType: "outgoing", ContentType: "text", SenderType: "user", SenderID: &user.ID,
|
||
}
|
||
require.NoError(t, db.Create(msg1).Error)
|
||
require.NoError(t, db.Create(msg2).Error)
|
||
|
||
// 正常路径：查询会话消息列表
|
||
messages, total, err := svc.ListByConversation(ctx, conv.ID, 0, 10)
|
||
assert.NoError(t, err)
|
||
assert.Equal(t, int64(2), total)
|
||
assert.Len(t, messages, 2)
|
||
|
||
// 正常路径：分页查询
|
||
messages2, total2, err2 := svc.ListByConversation(ctx, conv.ID, 1, 1)
|
||
assert.NoError(t, err2)
|
||
assert.Equal(t, int64(2), total2)
|
||
assert.Len(t, messages2, 1)
|
||
|
||
// 正常路径：空会话
|
||
messages3, total3, err3 := svc.ListByConversation(ctx, 9999, 0, 10)
|
||
assert.NoError(t, err3)
|
||
assert.Equal(t, int64(0), total3)
|
||
assert.Len(t, messages3, 0)
|
||
}
|
||
|
||
// ========== GetByID 测试 ==========
|
||
|
||
func TestMessageService_GetByID(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
ctx := context.Background()
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "Hello", MessageType: "incoming", ContentType: "text", SenderType: "contact",
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
|
||
// 正常路径：按ID查询
|
||
found, err := svc.GetByID(ctx, msg.ID)
|
||
assert.NoError(t, err)
|
||
assert.Equal(t, "Hello", found.Content)
|
||
assert.Equal(t, "incoming", found.MessageType)
|
||
|
||
// 错误路径：不存在的ID
|
||
_, err = svc.GetByID(ctx, 9999)
|
||
assert.Error(t, err)
|
||
}
|
||
|
||
// ========== GetByAccountAndID 测试 ==========
|
||
|
||
func TestMessageService_GetByAccountAndID(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
ctx := context.Background()
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "ScopedMsg", MessageType: "outgoing", ContentType: "text", SenderType: "user",
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
|
||
// 正常路径：按account和ID查询
|
||
found, err := svc.GetByAccountAndID(ctx, account.ID, msg.ID)
|
||
assert.NoError(t, err)
|
||
assert.Equal(t, "ScopedMsg", found.Content)
|
||
|
||
// 错误路径：accountID不匹配
|
||
_, err = svc.GetByAccountAndID(ctx, 9999, msg.ID)
|
||
assert.Error(t, err)
|
||
|
||
// 错误路径：ID不存在
|
||
_, err = svc.GetByAccountAndID(ctx, account.ID, 9999)
|
||
assert.Error(t, err)
|
||
}
|
||
|
||
// ========== GetByConversationAndID 测试 ==========
|
||
|
||
func TestMessageService_GetByConversationAndID(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
ctx := context.Background()
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "ConvScopedMsg", MessageType: "incoming", ContentType: "text", SenderType: "contact",
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
|
||
// 正常路径：按conversation和ID查询
|
||
found, err := svc.GetByConversationAndID(ctx, conv.ID, msg.ID)
|
||
assert.NoError(t, err)
|
||
assert.Equal(t, "ConvScopedMsg", found.Content)
|
||
|
||
// 错误路径：conversationID不匹配
|
||
_, err = svc.GetByConversationAndID(ctx, 9999, msg.ID)
|
||
assert.Error(t, err)
|
||
|
||
// 错误路径：ID不存在
|
||
_, err = svc.GetByConversationAndID(ctx, conv.ID, 9999)
|
||
assert.Error(t, err)
|
||
}
|
||
|
||
// ========== Search 测试 ==========
|
||
|
||
func TestMessageService_Search(t *testing.T) {
|
||
// Search底层使用ILIKE，SQLite不支持该语法
|
||
// 在SQLite环境下Search会返回SQL错误，Service层只做委托调用，这是预期行为
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
ctx := context.Background()
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg1 := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "important meeting", MessageType: "incoming", ContentType: "text", SenderType: "contact",
|
||
}
|
||
msg2 := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "casual chat", MessageType: "incoming", ContentType: "text", SenderType: "contact",
|
||
}
|
||
require.NoError(t, db.Create(msg1).Error)
|
||
require.NoError(t, db.Create(msg2).Error)
|
||
|
||
// ILIKE不兼容SQLite，在SQLite下搜索会返回SQL语法错误
|
||
// 这属于Repo层SQL兼容性问题，Service层本身只做委托调用
|
||
results, total, err := svc.Search(ctx, account.ID, "meeting", 0, 10, search.SearchModeILike)
|
||
if err != nil {
|
||
// SQLite环境：ILIKE语法不被支持，Search返回错误属于预期行为
|
||
t.Logf("Search在SQLite下返回错误（ILIKE不兼容）: %v", err)
|
||
} else {
|
||
// PostgreSQL环境：正常搜索
|
||
assert.Equal(t, int64(1), total)
|
||
assert.Len(t, results, 1)
|
||
if len(results) > 0 {
|
||
assert.Equal(t, "important meeting", results[0].Content)
|
||
}
|
||
}
|
||
}
|
||
|
||
// ========== Create 测试 ==========
|
||
|
||
func TestMessageService_Create(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
ctx := context.Background()
|
||
account := createTestAccount(t, db)
|
||
user := createTestUser(t, db, account.ID)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
// 正常路径：创建消息
|
||
req := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
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)
|
||
assert.Equal(t, conv.ID, created.ConversationID)
|
||
assert.Equal(t, account.ID, created.AccountID)
|
||
assert.Equal(t, "测试消息内容", created.Content)
|
||
assert.Equal(t, "outgoing", created.MessageType)
|
||
assert.Equal(t, "text", created.ContentType)
|
||
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)
|
||
require.NotNil(t, updatedConversation.LastMessageAt)
|
||
require.NotNil(t, updatedConversation.LastNonSysMsgAt)
|
||
assert.Equal(t, created.CreatedAt.Unix(), *updatedConversation.LastActivityAt)
|
||
assert.Equal(t, created.CreatedAt.Unix(), *updatedConversation.LastMessageAt)
|
||
assert.Equal(t, created.CreatedAt.Unix(), *updatedConversation.LastNonSysMsgAt)
|
||
|
||
// 正常路径：默认ContentType为text
|
||
reqNoContentType := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "无ContentType",
|
||
MessageType: "outgoing",
|
||
}
|
||
created2, err2 := svc.Create(ctx, account.ID, user.ID, reqNoContentType)
|
||
assert.NoError(t, err2)
|
||
assert.Equal(t, "text", created2.ContentType)
|
||
|
||
// Chatwoot parity: incoming messages are only allowed in API inboxes
|
||
reqIncomingNonAPI := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "非 API 收件箱 incoming",
|
||
MessageType: "incoming",
|
||
}
|
||
_, errIncomingNonAPI := svc.Create(ctx, account.ID, user.ID, reqIncomingNonAPI)
|
||
assert.EqualError(t, errIncomingNonAPI, "Incoming messages are only allowed in Api inboxes")
|
||
|
||
apiInbox := createTestInbox(t, db, account.ID, string(model.InboxChannelTypeAPI))
|
||
apiConv := createTestConversation(t, db, account.ID, apiInbox.ID, contact.ID)
|
||
reqIncomingAPI := CreateMessageRequest{
|
||
ConversationID: apiConv.ID,
|
||
Content: "API incoming",
|
||
MessageType: "incoming",
|
||
}
|
||
createdIncomingAPI, errIncomingAPI := svc.Create(ctx, account.ID, user.ID, reqIncomingAPI)
|
||
assert.NoError(t, errIncomingAPI)
|
||
assert.Equal(t, "incoming", createdIncomingAPI.MessageType)
|
||
|
||
// Chatwoot parity: campaign_id and template_params are stored in additional_attributes
|
||
templateParams := datatypes.JSON([]byte(`{"name":"customer_name","processed_params":{"body":"Hi"}}`))
|
||
reqWithAdditionalAttributes := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "带模板参数",
|
||
MessageType: "outgoing",
|
||
CampaignID: float64(42),
|
||
TemplateParams: templateParams,
|
||
}
|
||
createdWithAdditionalAttributes, errWithAdditionalAttributes := svc.Create(ctx, account.ID, user.ID, reqWithAdditionalAttributes)
|
||
assert.NoError(t, errWithAdditionalAttributes)
|
||
var additionalAttributes map[string]any
|
||
require.NoError(t, json.Unmarshal(createdWithAdditionalAttributes.AdditionalAttributes, &additionalAttributes))
|
||
assert.Equal(t, float64(42), additionalAttributes["campaign_id"])
|
||
assert.Equal(t, "customer_name", additionalAttributes["template_params"].(map[string]any)["name"])
|
||
|
||
// Chatwoot parity: email inbox recipient params are stored in content_attributes
|
||
emailInbox := createTestInbox(t, db, account.ID, string(model.InboxChannelTypeEmail))
|
||
emailConv := createTestConversation(t, db, account.ID, emailInbox.ID, contact.ID)
|
||
reqEmailRecipients := CreateMessageRequest{
|
||
ConversationID: emailConv.ID,
|
||
Content: "Email reply",
|
||
MessageType: "outgoing",
|
||
EmailHTMLContent: "<p>Email reply</p>",
|
||
ContentAttributes: datatypes.JSON([]byte(`{"email":{"html_content":{"reply":"<p>Hello</p>"}}}`)),
|
||
CCEmails: "agent_cc1@example.com, agent_cc2@example.com",
|
||
BCCEmails: "agent_bcc@example.com",
|
||
ToEmails: "customer@example.com",
|
||
}
|
||
createdEmailMessage, errEmailMessage := svc.Create(ctx, account.ID, user.ID, reqEmailRecipients)
|
||
assert.NoError(t, errEmailMessage)
|
||
var emailContentAttributes map[string]any
|
||
require.NoError(t, json.Unmarshal(createdEmailMessage.ContentAttributes, &emailContentAttributes))
|
||
assert.Equal(t, []any{"agent_cc1@example.com", "agent_cc2@example.com"}, emailContentAttributes["cc_emails"])
|
||
assert.Equal(t, []any{"agent_bcc@example.com"}, emailContentAttributes["bcc_emails"])
|
||
assert.Equal(t, []any{"customer@example.com"}, emailContentAttributes["to_emails"])
|
||
emailAttrs := emailContentAttributes["email"].(map[string]any)
|
||
assert.Equal(t, "Email reply", emailAttrs["text_content"].(map[string]any)["reply"])
|
||
assert.Equal(t, "<p>Email reply</p>", emailAttrs["html_content"].(map[string]any)["reply"])
|
||
|
||
// Chatwoot parity: stringified content_attributes are parsed as JSON objects
|
||
encodedAttrs, err := json.Marshal(`{"automation_rule_id":123,"items":[{"title":"Plan","value":"pro"}]}`)
|
||
require.NoError(t, err)
|
||
reqStringContentAttributes := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "String attrs",
|
||
MessageType: "outgoing",
|
||
ExternalCreatedAt: "2026-06-09 10:30:00 UTC",
|
||
ContentAttributes: datatypes.JSON(encodedAttrs),
|
||
}
|
||
createdStringAttrsMessage, errStringAttrsMessage := svc.Create(ctx, account.ID, user.ID, reqStringContentAttributes)
|
||
assert.NoError(t, errStringAttrsMessage)
|
||
var stringContentAttributes map[string]any
|
||
require.NoError(t, json.Unmarshal(createdStringAttrsMessage.ContentAttributes, &stringContentAttributes))
|
||
assert.Equal(t, float64(123), stringContentAttributes["automation_rule_id"])
|
||
assert.Equal(t, "2026-06-09 10:30:00 UTC", stringContentAttributes["external_created_at"])
|
||
require.Len(t, stringContentAttributes["items"], 1)
|
||
|
||
// Chatwoot parity: in_reply_to is normalized to a message in the same conversation
|
||
quotedSource := "provider-reply-1"
|
||
quotedMessage := &model.Message{
|
||
ConversationID: conv.ID,
|
||
AccountID: account.ID,
|
||
InboxID: inbox.ID,
|
||
Content: "Quoted",
|
||
MessageType: "incoming",
|
||
ContentType: "text",
|
||
SenderType: "contact",
|
||
SourceID: quotedSource,
|
||
}
|
||
require.NoError(t, db.Create(quotedMessage).Error)
|
||
reqReplyContentAttributes := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "Reply",
|
||
MessageType: "outgoing",
|
||
ContentAttributes: datatypes.JSON([]byte(fmt.Sprintf(`{"in_reply_to":%d,"items":[{"title":"Plan"}]}`, quotedMessage.ID))),
|
||
}
|
||
createdReplyMessage, errReplyMessage := svc.Create(ctx, account.ID, user.ID, reqReplyContentAttributes)
|
||
assert.NoError(t, errReplyMessage)
|
||
var replyContentAttributes map[string]any
|
||
require.NoError(t, json.Unmarshal(createdReplyMessage.ContentAttributes, &replyContentAttributes))
|
||
assert.Equal(t, float64(quotedMessage.ID), replyContentAttributes["in_reply_to"])
|
||
assert.Equal(t, quotedSource, replyContentAttributes["in_reply_to_external_id"])
|
||
require.Len(t, replyContentAttributes["items"], 1)
|
||
|
||
otherConv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
reqCrossConversationReply := CreateMessageRequest{
|
||
ConversationID: otherConv.ID,
|
||
Content: "Cross reply",
|
||
MessageType: "outgoing",
|
||
ContentAttributes: datatypes.JSON([]byte(fmt.Sprintf(`{"in_reply_to":%d}`, quotedMessage.ID))),
|
||
}
|
||
createdCrossReplyMessage, errCrossReplyMessage := svc.Create(ctx, account.ID, user.ID, reqCrossConversationReply)
|
||
assert.NoError(t, errCrossReplyMessage)
|
||
var crossReplyContentAttributes map[string]any
|
||
require.NoError(t, json.Unmarshal(createdCrossReplyMessage.ContentAttributes, &crossReplyContentAttributes))
|
||
assert.Nil(t, crossReplyContentAttributes["in_reply_to"])
|
||
assert.Nil(t, crossReplyContentAttributes["in_reply_to_external_id"])
|
||
|
||
// Chatwoot parity: structured bot content types can be created without text content
|
||
for _, contentType := range []string{"input_select", "cards"} {
|
||
createdStructuredMessage, errStructuredMessage := svc.Create(ctx, account.ID, user.ID, CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
MessageType: "outgoing",
|
||
ContentType: contentType,
|
||
ContentAttributes: datatypes.JSON([]byte(`{"items":[{"title":"Plan","value":"pro"}]}`)),
|
||
})
|
||
assert.NoError(t, errStructuredMessage)
|
||
assert.Equal(t, contentType, createdStructuredMessage.ContentType)
|
||
assert.Empty(t, createdStructuredMessage.Content)
|
||
}
|
||
|
||
// Chatwoot parity: sender_type=AgentBot uses an accessible agent bot as sender
|
||
require.NoError(t, db.AutoMigrate(&model.AgentBot{}))
|
||
agentBot := &model.AgentBot{AccountID: &account.ID, Name: "Reply Bot", BotType: "webhook"}
|
||
require.NoError(t, db.Create(agentBot).Error)
|
||
require.NoError(t, db.Model(&conv).Update("assignee_agent_bot_id", agentBot.ID).Error)
|
||
reqAgentBotSender := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "Agent bot reply",
|
||
MessageType: "outgoing",
|
||
SenderType: string(model.SenderTypeAgentBot),
|
||
SenderID: agentBot.ID,
|
||
}
|
||
createdAgentBotMessage, errAgentBotMessage := svc.Create(ctx, account.ID, user.ID, reqAgentBotSender)
|
||
assert.NoError(t, errAgentBotMessage)
|
||
assert.Equal(t, string(model.SenderTypeAgentBot), createdAgentBotMessage.SenderType)
|
||
assert.Equal(t, agentBot.ID, *createdAgentBotMessage.SenderID)
|
||
|
||
reqMissingAgentBotSender := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "Fallback reply",
|
||
MessageType: "outgoing",
|
||
SenderType: string(model.SenderTypeAgentBot),
|
||
SenderID: agentBot.ID + 999,
|
||
}
|
||
createdFallbackMessage, errFallbackMessage := svc.Create(ctx, account.ID, user.ID, reqMissingAgentBotSender)
|
||
assert.Nil(t, createdFallbackMessage)
|
||
assert.EqualError(t, errFallbackMessage, "agent bot not found")
|
||
|
||
// Chatwoot parity: is_voice_message tags only audio attachments
|
||
reqVoiceMessage := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "voice note",
|
||
MessageType: "outgoing",
|
||
IsVoiceMessage: true,
|
||
Attachments: []MessageAttachmentInput{
|
||
{FileName: "voice.mp3", FileSize: 1234, ContentType: "audio/mpeg"},
|
||
{FileName: "transcript.txt", FileSize: 12, ContentType: "text/plain"},
|
||
},
|
||
}
|
||
createdVoiceMessage, errVoiceMessage := svc.Create(ctx, account.ID, user.ID, reqVoiceMessage)
|
||
assert.NoError(t, errVoiceMessage)
|
||
var voiceAttachments []model.Attachment
|
||
require.NoError(t, db.Where("message_id = ?", createdVoiceMessage.ID).Order("id ASC").Find(&voiceAttachments).Error)
|
||
require.Len(t, voiceAttachments, 2)
|
||
assert.Equal(t, `{"is_voice_message":true}`, voiceAttachments[0].Metadata)
|
||
assert.Empty(t, voiceAttachments[1].Metadata)
|
||
|
||
// 正常路径：创建私密消息
|
||
reqPrivate := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "私密消息",
|
||
MessageType: "private_note",
|
||
Private: true,
|
||
}
|
||
created3, err3 := svc.Create(ctx, account.ID, user.ID, reqPrivate)
|
||
assert.NoError(t, err3)
|
||
assert.True(t, created3.Private)
|
||
|
||
// 错误路径：验证失败 - 空Content
|
||
reqEmpty := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "",
|
||
MessageType: "outgoing",
|
||
}
|
||
_, err4 := svc.Create(ctx, account.ID, user.ID, reqEmpty)
|
||
assert.Error(t, err4)
|
||
|
||
// 错误路径：验证失败 - 无效MessageType
|
||
reqInvalid := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "内容",
|
||
MessageType: "invalid_type",
|
||
}
|
||
_, err5 := svc.Create(ctx, account.ID, user.ID, reqInvalid)
|
||
assert.Error(t, err5)
|
||
|
||
// 错误路径：验证失败 - 无效ContentType
|
||
reqInvalidCT := CreateMessageRequest{
|
||
ConversationID: conv.ID,
|
||
Content: "内容",
|
||
MessageType: "outgoing",
|
||
ContentType: "invalid_ct",
|
||
}
|
||
_, err6 := svc.Create(ctx, account.ID, user.ID, reqInvalidCT)
|
||
assert.Error(t, err6)
|
||
|
||
// 错误路径：验证失败 - 缺少ConversationID
|
||
reqNoConvID := CreateMessageRequest{
|
||
Content: "内容",
|
||
MessageType: "outgoing",
|
||
}
|
||
_, err7 := svc.Create(ctx, account.ID, user.ID, reqNoConvID)
|
||
assert.Error(t, err7)
|
||
}
|
||
|
||
func TestMessageService_ShangwutongImportIsIdempotentAndDoesNotQueueOutbound(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
wp := worker.NewWorkerPool(db)
|
||
svc.SetWorkerPool(wp)
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "shangwutong")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
req := CreateMessageRequest{
|
||
ConversationID: conversation.ID, MessageType: "incoming", ContentType: "text", Content: "您好",
|
||
SourceID: fmt.Sprintf("swt:%d:visitor:2:98765:0", inbox.ID), External: true,
|
||
ExternalCreatedAt: time.Date(2026, 8, 1, 9, 0, 0, 0, time.UTC).Format(time.RFC3339Nano),
|
||
ExternalSourceIDs: datatypes.JSON([]byte(`{"shangwutong":"98765"}`)),
|
||
ContentAttributes: datatypes.JSON([]byte(`{"swt":{"kind":2,"seq_id":98765}}`)),
|
||
AdditionalAttributes: datatypes.JSON([]byte(`{"senderName":"商务通访客"}`)),
|
||
}
|
||
created, err := svc.Create(context.Background(), account.ID, 0, req)
|
||
require.NoError(t, err)
|
||
replayed, err := svc.Create(context.Background(), account.ID, 0, req)
|
||
require.NoError(t, err)
|
||
require.Equal(t, created.ID, replayed.ID)
|
||
require.True(t, replayed.IdempotentReplay)
|
||
require.True(t, created.External)
|
||
require.Equal(t, string(model.SenderTypeContact), created.SenderType)
|
||
require.NotNil(t, created.SenderID)
|
||
require.Equal(t, contact.ID, *created.SenderID)
|
||
var count int64
|
||
require.NoError(t, db.Model(&model.Message{}).Where("inbox_id = ? AND source_id = ?", inbox.ID, req.SourceID).Count(&count).Error)
|
||
require.EqualValues(t, 1, count)
|
||
require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ?", TaskTypeShangwutongWebhookDelivery).Count(&count).Error)
|
||
require.Zero(t, count)
|
||
req.Content = "不同正文"
|
||
_, err = svc.Create(context.Background(), account.ID, 0, req)
|
||
require.ErrorIs(t, err, ErrMessageIdempotencyConflict)
|
||
}
|
||
|
||
func TestMessageServiceResolveConversationForRouteDoesNotFallBackToInternalID(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
displayID := conversation.ID + 1000
|
||
require.NoError(t, db.Model(conversation).Update("display_id", displayID).Error)
|
||
|
||
resolved, err := svc.ResolveConversationForRoute(context.Background(), account.ID, displayID)
|
||
require.NoError(t, err)
|
||
require.Equal(t, conversation.ID, resolved.ID)
|
||
|
||
_, err = svc.ResolveConversationForRoute(context.Background(), account.ID, conversation.ID)
|
||
require.ErrorIs(t, err, gorm.ErrRecordNotFound)
|
||
}
|
||
|
||
func TestMessageService_ShangwutongOutboundResultAndRetryStayDurable(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
wp := worker.NewWorkerPool(db)
|
||
svc.SetWorkerPool(wp)
|
||
account := createTestAccount(t, db)
|
||
user := createTestUser(t, db, account.ID)
|
||
inbox := createTestInbox(t, db, account.ID, "shangwutong")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
message, err := svc.Create(context.Background(), account.ID, user.ID, CreateMessageRequest{
|
||
ConversationID: conversation.ID, MessageType: "outgoing", ContentType: "text", Content: "回复",
|
||
ContentAttributes: datatypes.JSON([]byte(`{"business":"keep"}`)),
|
||
})
|
||
require.NoError(t, err)
|
||
require.Equal(t, string(model.MessageStatusProgress), message.Status)
|
||
var jobs []model.BackgroundJob
|
||
require.NoError(t, db.Where("job_type = ?", TaskTypeShangwutongWebhookDelivery).Order("id ASC").Find(&jobs).Error)
|
||
require.Len(t, jobs, 1)
|
||
require.Equal(t, fmt.Sprintf("api-inbox-message:%d:created", message.ID), jobs[0].IdempotencyKey)
|
||
|
||
uncertain := ShangwutongMessageResult{ResultVersion: 1, Status: "uncertain", OccurredAt: time.Date(2026, 8, 1, 9, 1, 0, 0, time.UTC)}
|
||
updated, applied, err := svc.ApplyShangwutongMessageResult(context.Background(), account.ID, inbox.ID, message.ID, uncertain)
|
||
require.NoError(t, err)
|
||
require.True(t, applied)
|
||
require.Equal(t, string(model.MessageStatusProgress), updated.Status)
|
||
require.Contains(t, string(updated.ContentAttributes), `"external_delivery_state":"uncertain"`)
|
||
_, applied, err = svc.ApplyShangwutongMessageResult(context.Background(), account.ID, inbox.ID, message.ID, uncertain)
|
||
require.NoError(t, err)
|
||
require.False(t, applied)
|
||
conflict := uncertain
|
||
conflict.Status = "failed"
|
||
errorCode := "uncertain_timeout"
|
||
conflict.ErrorCode = &errorCode
|
||
_, _, err = svc.ApplyShangwutongMessageResult(context.Background(), account.ID, inbox.ID, message.ID, conflict)
|
||
require.ErrorIs(t, err, ErrShangwutongMessageResultConflict)
|
||
|
||
failed := ShangwutongMessageResult{ResultVersion: 2, Status: "failed", ErrorCode: &errorCode, OccurredAt: time.Date(2026, 8, 1, 9, 2, 0, 0, time.UTC)}
|
||
updated, applied, err = svc.ApplyShangwutongMessageResult(context.Background(), account.ID, inbox.ID, message.ID, failed)
|
||
require.NoError(t, err)
|
||
require.True(t, applied)
|
||
require.Equal(t, string(model.MessageStatusFailed), updated.Status)
|
||
|
||
retried, err := svc.RetryInConversation(context.Background(), account.ID, conversation.ID, message.ID)
|
||
require.NoError(t, err)
|
||
require.Equal(t, string(model.MessageStatusProgress), retried.Status)
|
||
attrs := map[string]any{}
|
||
require.NoError(t, json.Unmarshal(retried.ContentAttributes, &attrs))
|
||
require.Equal(t, "keep", attrs["business"])
|
||
require.EqualValues(t, 1, attrs["external_retry_version"])
|
||
require.NotContains(t, attrs, "external_error")
|
||
require.NoError(t, db.Where("job_type = ?", TaskTypeShangwutongWebhookDelivery).Order("id ASC").Find(&jobs).Error)
|
||
require.Len(t, jobs, 2)
|
||
require.Equal(t, fmt.Sprintf("api-inbox-message:%d:retry:1", message.ID), jobs[1].IdempotencyKey)
|
||
|
||
externalID := "98766"
|
||
sent := ShangwutongMessageResult{ResultVersion: 3, Status: "sent", ExternalID: &externalID, OccurredAt: time.Date(2026, 8, 1, 9, 3, 0, 0, time.UTC)}
|
||
updated, applied, err = svc.ApplyShangwutongMessageResult(context.Background(), account.ID, inbox.ID, message.ID, sent)
|
||
require.NoError(t, err)
|
||
require.True(t, applied)
|
||
require.Equal(t, string(model.MessageStatusSent), updated.Status)
|
||
require.JSONEq(t, `{"shangwutong":"98766"}`, string(updated.ExternalSourceIDs))
|
||
|
||
secondSent := ShangwutongMessageResult{ResultVersion: 4, Status: "sent", ExternalIDs: []string{"98766", "98767"}, OccurredAt: time.Date(2026, 8, 1, 9, 4, 0, 0, time.UTC)}
|
||
updated, applied, err = svc.ApplyShangwutongMessageResult(context.Background(), account.ID, inbox.ID, message.ID, secondSent)
|
||
require.NoError(t, err)
|
||
require.True(t, applied)
|
||
require.JSONEq(t, `{"shangwutong":["98766","98767"]}`, string(updated.ExternalSourceIDs))
|
||
}
|
||
|
||
func TestMessageServiceRejectsUnassignedAgentBotReply(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
require.NoError(t, db.AutoMigrate(&model.AgentBot{}))
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "api")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
bot := &model.AgentBot{AccountID: &account.ID, Name: "Bot", BotType: "webhook"}
|
||
require.NoError(t, db.Create(bot).Error)
|
||
|
||
_, err := svc.Create(context.Background(), account.ID, bot.ID, CreateMessageRequest{
|
||
ConversationID: conversation.ID, MessageType: "outgoing", ContentType: "text", Content: "late reply",
|
||
SenderType: string(model.SenderTypeAgentBot), SenderID: bot.ID,
|
||
})
|
||
require.EqualError(t, err, "agent bot is not assigned to this conversation")
|
||
}
|
||
|
||
func TestMessageService_ConversationScopedMessageActions(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
ctx := context.Background()
|
||
account := createTestAccount(t, db)
|
||
user := createTestUser(t, db, account.ID)
|
||
inbox := createTestInbox(t, db, account.ID, "api")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
otherConversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
message := &model.Message{
|
||
ConversationID: conversation.ID,
|
||
AccountID: account.ID,
|
||
InboxID: inbox.ID,
|
||
SenderID: &user.ID,
|
||
SenderType: "user",
|
||
Content: "route scoped",
|
||
MessageType: "outgoing",
|
||
ContentType: "text",
|
||
Status: "failed",
|
||
}
|
||
require.NoError(t, db.Create(message).Error)
|
||
|
||
found, err := svc.GetByAccountConversationAndID(ctx, account.ID, conversation.ID, message.ID)
|
||
require.NoError(t, err)
|
||
assert.Equal(t, message.ID, found.ID)
|
||
|
||
_, err = svc.GetByAccountConversationAndID(ctx, account.ID, otherConversation.ID, message.ID)
|
||
assert.Error(t, err)
|
||
|
||
_, err = svc.UpdateInConversation(ctx, account.ID, otherConversation.ID, message.ID, UpdateMessageRequest{Status: "delivered"})
|
||
assert.Error(t, err)
|
||
require.NoError(t, db.First(message, message.ID).Error)
|
||
assert.Equal(t, "route scoped", message.Content)
|
||
assert.Equal(t, "failed", message.Status)
|
||
|
||
updated, err := svc.UpdateInConversation(ctx, account.ID, conversation.ID, message.ID, UpdateMessageRequest{Status: "delivered"})
|
||
require.NoError(t, err)
|
||
assert.Equal(t, "route scoped", updated.Content)
|
||
assert.Equal(t, "delivered", updated.Status)
|
||
require.NoError(t, db.Model(message).Update("status", "failed").Error)
|
||
|
||
_, err = svc.RetryInConversation(ctx, account.ID, otherConversation.ID, message.ID)
|
||
assert.Error(t, err)
|
||
require.NoError(t, db.First(message, message.ID).Error)
|
||
assert.Equal(t, "failed", message.Status)
|
||
|
||
retried, err := svc.RetryInConversation(ctx, account.ID, conversation.ID, message.ID)
|
||
require.NoError(t, err)
|
||
assert.Equal(t, "sent", retried.Status)
|
||
|
||
_, err = svc.DeleteInConversation(ctx, account.ID, otherConversation.ID, message.ID)
|
||
assert.Error(t, err)
|
||
require.NoError(t, db.First(message, message.ID).Error)
|
||
assert.NotContains(t, string(message.ContentAttributes), "deleted")
|
||
|
||
deleted, err := svc.DeleteInConversation(ctx, account.ID, conversation.ID, message.ID)
|
||
require.NoError(t, err)
|
||
assert.Equal(t, "This message was deleted", deleted.Content)
|
||
assert.JSONEq(t, `{"deleted":true}`, string(deleted.ContentAttributes))
|
||
}
|
||
|
||
// ========== Update 测试 ==========
|
||
|
||
func TestMessageService_Update(t *testing.T) {
|
||
db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
ctx := context.Background()
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "api")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "原始内容", MessageType: "outgoing", ContentType: "text", SenderType: "user", Status: "sent",
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
listener := &mockRetryListener{}
|
||
dispatcher.Register(listener)
|
||
|
||
req := UpdateMessageRequest{Status: "failed", ExternalError: "provider rejected message"}
|
||
updated, err := svc.Update(ctx, account.ID, msg.ID, req)
|
||
assert.NoError(t, err)
|
||
assert.Equal(t, "failed", updated.Status)
|
||
assert.Equal(t, "原始内容", updated.Content)
|
||
assert.JSONEq(t, `{"external_error":"provider rejected message"}`, string(updated.ContentAttributes))
|
||
assert.True(t, listener.received)
|
||
assert.Equal(t, channel.EventMessageUpdated, listener.lastEventType)
|
||
assert.Equal(t, account.ID, listener.lastAccountID)
|
||
assert.Equal(t, inbox.ID, listener.lastInboxID)
|
||
assert.Equal(t, conv.ID, listener.lastConvID)
|
||
eventMessage, ok := listener.lastData["message"].(*model.Message)
|
||
require.True(t, ok)
|
||
assert.Equal(t, msg.ID, eventMessage.ID)
|
||
assert.Equal(t, "failed", eventMessage.Status)
|
||
assert.JSONEq(t, `{"external_error":"provider rejected message"}`, string(eventMessage.ContentAttributes))
|
||
|
||
reqEmpty := UpdateMessageRequest{}
|
||
updated2, err2 := svc.Update(ctx, account.ID, msg.ID, reqEmpty)
|
||
assert.NoError(t, err2)
|
||
assert.Equal(t, "failed", updated2.Status)
|
||
assert.Equal(t, "原始内容", updated2.Content)
|
||
|
||
_, err = svc.Update(ctx, account.ID, msg.ID, UpdateMessageRequest{Status: "invalid"})
|
||
assert.Error(t, err)
|
||
|
||
webInbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
webConv := createTestConversation(t, db, account.ID, webInbox.ID, contact.ID)
|
||
webMessage := &model.Message{
|
||
ConversationID: webConv.ID, AccountID: account.ID, InboxID: webInbox.ID,
|
||
Content: "web", MessageType: "outgoing", ContentType: "text", SenderType: "user", Status: "sent",
|
||
}
|
||
require.NoError(t, db.Create(webMessage).Error)
|
||
_, err = svc.Update(ctx, account.ID, webMessage.ID, UpdateMessageRequest{Status: "delivered"})
|
||
assert.Error(t, err)
|
||
|
||
// 错误路径：accountID不匹配
|
||
_, err = svc.Update(ctx, 9999, msg.ID, req)
|
||
assert.Error(t, err)
|
||
|
||
// 错误路径：消息ID不存在
|
||
_, err = svc.Update(ctx, account.ID, 9999, req)
|
||
assert.Error(t, err)
|
||
}
|
||
|
||
// ========== Delete 测试 ==========
|
||
|
||
func TestMessageService_Delete(t *testing.T) {
|
||
db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
ctx := context.Background()
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "待删除消息", MessageType: "incoming", ContentType: "text", SenderType: "contact",
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
attachment := &model.Attachment{MessageID: msg.ID, AccountID: account.ID, FileType: "file", FileName: "delete.txt"}
|
||
require.NoError(t, db.Create(attachment).Error)
|
||
listener := &mockRetryListener{}
|
||
dispatcher.Register(listener)
|
||
|
||
// 正常路径：Chatwoot 删除会保留消息并标记 content_attributes.deleted
|
||
deleted, err := svc.Delete(ctx, account.ID, msg.ID)
|
||
assert.NoError(t, err)
|
||
assert.Equal(t, "This message was deleted", deleted.Content)
|
||
assert.True(t, listener.received)
|
||
assert.Equal(t, channel.EventMessageDeleted, listener.lastEventType)
|
||
assert.Equal(t, account.ID, listener.lastAccountID)
|
||
assert.Equal(t, inbox.ID, listener.lastInboxID)
|
||
assert.Equal(t, conv.ID, listener.lastConvID)
|
||
eventMessage, ok := listener.lastData["message"].(*model.Message)
|
||
require.True(t, ok)
|
||
assert.Equal(t, msg.ID, eventMessage.ID)
|
||
assert.Equal(t, "This message was deleted", eventMessage.Content)
|
||
assert.JSONEq(t, `{"deleted":true}`, string(eventMessage.ContentAttributes))
|
||
|
||
stored, err := svc.GetByAccountAndID(ctx, account.ID, msg.ID)
|
||
assert.NoError(t, err)
|
||
assert.JSONEq(t, `{"deleted":true}`, string(stored.ContentAttributes))
|
||
|
||
var attachmentCount int64
|
||
require.NoError(t, db.Model(&model.Attachment{}).Where("message_id = ?", msg.ID).Count(&attachmentCount).Error)
|
||
assert.Equal(t, int64(0), attachmentCount)
|
||
|
||
// 错误路径：accountID不匹配
|
||
msg2 := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "另一条消息", MessageType: "incoming", ContentType: "text", SenderType: "contact",
|
||
}
|
||
require.NoError(t, db.Create(msg2).Error)
|
||
_, err = svc.Delete(ctx, 9999, msg2.ID)
|
||
assert.Error(t, err)
|
||
|
||
// 错误路径：ID不存在
|
||
_, err = svc.Delete(ctx, account.ID, 9999)
|
||
assert.Error(t, err)
|
||
}
|
||
|
||
// ========== CountByConversation 测试 ==========
|
||
|
||
func TestMessageService_CountByConversation(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
ctx := context.Background()
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
// 创建3条消息
|
||
for i := 0; i < 3; i++ {
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: fmt.Sprintf("消息%c", 'A'+i),
|
||
MessageType: "incoming", ContentType: "text", SenderType: "contact",
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
}
|
||
|
||
// 正常路径：统计会话消息数量
|
||
count, err := svc.CountByConversation(ctx, conv.ID)
|
||
assert.NoError(t, err)
|
||
assert.Equal(t, int64(3), count)
|
||
|
||
// 正常路径：空会话
|
||
count2, err2 := svc.CountByConversation(ctx, 9999)
|
||
assert.NoError(t, err2)
|
||
assert.Equal(t, int64(0), count2)
|
||
}
|
||
|
||
// ========== Translate 测试 ==========
|
||
|
||
func TestMessageService_Translate(t *testing.T) {
|
||
ctx := context.Background()
|
||
|
||
t.Run("success", func(t *testing.T) {
|
||
mockLLM := &mockMessageLLMProvider{
|
||
chatResponse: &llm.ChatResponse{
|
||
Choices: []llm.ChatChoice{
|
||
{Message: llm.ChatMessage{Role: "assistant", Content: "Hello"}},
|
||
},
|
||
},
|
||
}
|
||
db, _, _, svc := setupMessageService(t, mockLLM)
|
||
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "你好", MessageType: "incoming", ContentType: "text", SenderType: "contact",
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
|
||
result, err := svc.Translate(ctx, account.ID, msg.ID, TranslateMessageRequest{TargetLanguage: "en"})
|
||
assert.NoError(t, err)
|
||
assert.Equal(t, msg.ID, result.ID)
|
||
assert.Equal(t, "你好", result.OriginalContent)
|
||
assert.Equal(t, "Hello", result.TranslatedContent)
|
||
assert.Equal(t, "en", result.TargetLanguage)
|
||
assert.Equal(t, 1, mockLLM.chatCalls)
|
||
|
||
resultCached, errCached := svc.Translate(ctx, account.ID, msg.ID, TranslateMessageRequest{TargetLanguage: "en"})
|
||
assert.NoError(t, errCached)
|
||
assert.Equal(t, "Hello", resultCached.TranslatedContent)
|
||
assert.Equal(t, 1, mockLLM.chatCalls)
|
||
})
|
||
|
||
t.Run("empty_choices", func(t *testing.T) {
|
||
mockLLM := &mockMessageLLMProvider{
|
||
chatResponse: &llm.ChatResponse{
|
||
Choices: []llm.ChatChoice{},
|
||
},
|
||
}
|
||
db, _, _, svc := setupMessageService(t, mockLLM)
|
||
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "你好", MessageType: "incoming", ContentType: "text", SenderType: "contact",
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
|
||
result, err := svc.Translate(ctx, account.ID, msg.ID, TranslateMessageRequest{TargetLanguage: "en"})
|
||
assert.NoError(t, err)
|
||
assert.Equal(t, "", result.TranslatedContent)
|
||
})
|
||
|
||
t.Run("llm_error", func(t *testing.T) {
|
||
mockLLM := &mockMessageLLMProvider{
|
||
chatErr: fmt.Errorf("LLM service unavailable"),
|
||
}
|
||
db, _, _, svc := setupMessageService(t, mockLLM)
|
||
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "你好", MessageType: "incoming", ContentType: "text", SenderType: "contact",
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
|
||
_, err := svc.Translate(ctx, account.ID, msg.ID, TranslateMessageRequest{TargetLanguage: "en"})
|
||
assert.Error(t, err)
|
||
})
|
||
|
||
t.Run("message_not_found", func(t *testing.T) {
|
||
mockLLM := &mockMessageLLMProvider{
|
||
chatResponse: &llm.ChatResponse{
|
||
Choices: []llm.ChatChoice{
|
||
{Message: llm.ChatMessage{Role: "assistant", Content: "translated"}},
|
||
},
|
||
},
|
||
}
|
||
db, _, _, svc := setupMessageService(t, mockLLM)
|
||
|
||
account := createTestAccount(t, db)
|
||
|
||
_, err := svc.Translate(ctx, account.ID, 9999, TranslateMessageRequest{TargetLanguage: "en"})
|
||
assert.Error(t, err)
|
||
})
|
||
|
||
t.Run("validation_empty_target_language", func(t *testing.T) {
|
||
mockLLM := &mockMessageLLMProvider{}
|
||
db, _, _, svc := setupMessageService(t, mockLLM)
|
||
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "你好", MessageType: "incoming", ContentType: "text", SenderType: "contact",
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
|
||
_, err := svc.Translate(ctx, account.ID, msg.ID, TranslateMessageRequest{TargetLanguage: ""})
|
||
assert.Error(t, err)
|
||
})
|
||
}
|
||
|
||
// ========== Retry 测试 ==========
|
||
|
||
func TestMessageService_Retry(t *testing.T) {
|
||
ctx := context.Background()
|
||
|
||
t.Run("success", func(t *testing.T) {
|
||
db, _, dispatcher, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "failed message", MessageType: "outgoing", ContentType: "text", SenderType: "user",
|
||
Status: "failed", ContentAttributes: datatypes.JSON([]byte(`{"external_error":"provider failed"}`)),
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
|
||
// Register a listener to verify dispatch
|
||
listener := &mockRetryListener{}
|
||
dispatcher.Register(listener)
|
||
|
||
retried, err := svc.Retry(ctx, account.ID, msg.ID)
|
||
assert.NoError(t, err)
|
||
assert.Equal(t, "sent", retried.Status)
|
||
assert.JSONEq(t, `{}`, string(retried.ContentAttributes))
|
||
assert.True(t, listener.received)
|
||
assert.Equal(t, channel.EventMessageStatusUpdated, listener.lastEventType)
|
||
assert.Equal(t, account.ID, listener.lastAccountID)
|
||
assert.Equal(t, inbox.ID, listener.lastInboxID)
|
||
assert.Equal(t, conv.ID, listener.lastConvID)
|
||
assert.Equal(t, "sent", listener.lastData["status"])
|
||
assert.Equal(t, msg.ID, listener.lastData["message_id"])
|
||
|
||
var stored model.Message
|
||
require.NoError(t, db.First(&stored, msg.ID).Error)
|
||
assert.JSONEq(t, `{}`, string(stored.ContentAttributes))
|
||
})
|
||
|
||
t.Run("queues_send_reply_when_worker_configured", func(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
wp := worker.NewWorkerPool(db)
|
||
svc.SetWorkerPool(wp)
|
||
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "failed message", MessageType: "outgoing", ContentType: "text", SenderType: "user",
|
||
Status: "failed", ContentAttributes: datatypes.JSON([]byte(`{"external_error":"provider failed"}`)),
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
|
||
retried, err := svc.Retry(ctx, account.ID, msg.ID)
|
||
require.NoError(t, err)
|
||
assert.Equal(t, "sent", retried.Status)
|
||
|
||
var job model.BackgroundJob
|
||
require.NoError(t, db.Where("job_type = ?", TaskTypeMessageSendReply).First(&job).Error)
|
||
assert.Equal(t, model.BackgroundJobStatusQueued, job.Status)
|
||
assert.Equal(t, "message:send_reply", job.JobType)
|
||
assert.JSONEq(t, fmt.Sprintf(`{"message_id":%d}`, msg.ID), string(job.Payload))
|
||
})
|
||
|
||
t.Run("message_not_found", func(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
account := createTestAccount(t, db)
|
||
|
||
_, err := svc.Retry(ctx, account.ID, 9999)
|
||
assert.Error(t, err)
|
||
})
|
||
|
||
t.Run("account_mismatch", func(t *testing.T) {
|
||
db, _, _, svc := setupMessageServiceWithDefaultLLM(t)
|
||
|
||
account := createTestAccount(t, db)
|
||
inbox := createTestInbox(t, db, account.ID, "web_widget")
|
||
contact := createTestContact(t, db, account.ID)
|
||
conv := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
|
||
|
||
msg := &model.Message{
|
||
ConversationID: conv.ID, AccountID: account.ID, InboxID: inbox.ID,
|
||
Content: "test", MessageType: "outgoing", ContentType: "text", SenderType: "user",
|
||
Status: "failed",
|
||
}
|
||
require.NoError(t, db.Create(msg).Error)
|
||
|
||
_, err := svc.Retry(ctx, 9999, msg.ID)
|
||
assert.Error(t, err)
|
||
})
|
||
}
|