feat(conversations): finish message mutation parity

This commit is contained in:
2026-06-05 02:10:42 +08:00
parent 5c44ca0c50
commit a465bbe09d
10 changed files with 420 additions and 42 deletions
+139 -16
View File
@@ -2,6 +2,7 @@ package service
import (
"context"
"encoding/json"
"fmt"
"strings"
@@ -108,14 +109,21 @@ func (s *MessageService) Search(ctx context.Context, accountID uint, query strin
// CreateMessageRequest is the DTO for creating a message.
// Reference: Chatwoot app/controllers/api/v1/accounts/conversations/messages_controller.rb #create
type CreateMessageRequest struct {
ConversationID uint `json:"conversation_id" validate:"required"`
Content string `json:"content" validate:"required,min=1"`
MessageType string `json:"message_type,omitempty"`
ContentType string `json:"content_type,omitempty"`
Private bool `json:"private,omitempty"`
SourceID string `json:"source_id,omitempty"`
EchoID string `json:"echo_id,omitempty"`
ContentAttributes datatypes.JSON `json:"content_attributes,omitempty"`
ConversationID uint `json:"conversation_id" validate:"required"`
Content string `json:"content"`
MessageType string `json:"message_type,omitempty"`
ContentType string `json:"content_type,omitempty"`
Private bool `json:"private,omitempty"`
SourceID string `json:"source_id,omitempty"`
EchoID string `json:"echo_id,omitempty"`
ContentAttributes datatypes.JSON `json:"content_attributes,omitempty"`
Attachments []MessageAttachmentInput `json:"-"`
}
type MessageAttachmentInput struct {
FileName string
FileSize int
ContentType string
}
// Create creates a new message.
@@ -133,6 +141,9 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint
if err := pkgvalidator.ValidateStruct(req); err != nil {
return nil, err
}
if strings.TrimSpace(req.Content) == "" && len(req.Attachments) == 0 {
return nil, fmt.Errorf("content is required")
}
var conversation model.Conversation
if err := s.repo.DB().WithContext(ctx).Where("account_id = ? AND id = ?", accountID, req.ConversationID).First(&conversation).Error; err != nil {
@@ -163,7 +174,26 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint
}
}
if err := s.repo.Create(ctx, message); err != nil {
if err := s.repo.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Create(message).Error; err != nil {
return err
}
for _, input := range req.Attachments {
attachment := &model.Attachment{
MessageID: message.ID,
AccountID: accountID,
FileType: attachmentFileType(input.ContentType),
FileURL: attachmentDataURL(message.ID, input.FileName),
ThumbURL: attachmentThumbURL(input.ContentType, message.ID, input.FileName),
FileSize: input.FileSize,
FileName: input.FileName,
}
if err := tx.Create(attachment).Error; err != nil {
return err
}
}
return nil
}); err != nil {
applogger.L().Errorf("Failed to create message: %v", err)
return nil, err
}
@@ -219,7 +249,9 @@ func validContentType(value string) bool {
// UpdateMessageRequest is the DTO for updating a message.
type UpdateMessageRequest struct {
Content string `json:"content,omitempty" validate:"omitempty,min=1"`
Content string `json:"content,omitempty" validate:"omitempty,min=1"`
Status string `json:"status,omitempty"`
ExternalError string `json:"external_error,omitempty"`
}
// Update modifies an existing message.
@@ -236,6 +268,16 @@ func (s *MessageService) Update(ctx context.Context, accountID, id uint, req Upd
if req.Content != "" {
message.Content = req.Content
}
if req.Status != "" && validMessageStatus(req.Status) {
if !s.messageInboxIsAPI(ctx, message.InboxID) {
return nil, fmt.Errorf("Message status update is only allowed for API inboxes")
}
if message.Status == "read" && req.Status == "delivered" {
return message, nil
}
message.Status = req.Status
message.ContentAttributes = setMessageExternalError(message.ContentAttributes, req.Status, req.ExternalError)
}
if err := s.repo.Update(ctx, message); err != nil {
return nil, err
@@ -248,22 +290,47 @@ func (s *MessageService) Update(ctx context.Context, accountID, id uint, req Upd
return message, nil
}
// Delete soft-deletes a message and dispatches EventMessageDeleted.
func (s *MessageService) Delete(ctx context.Context, accountID, id uint) error {
func (s *MessageService) messageInboxIsAPI(ctx context.Context, inboxID uint) bool {
var inbox model.Inbox
if err := s.repo.DB().WithContext(ctx).First(&inbox, inboxID).Error; err != nil {
return false
}
switch strings.ToLower(inbox.ChannelType) {
case "api", "channel::api":
return true
default:
return false
}
}
// Delete marks a message deleted using Chatwoot's visible tombstone payload.
func (s *MessageService) Delete(ctx context.Context, accountID, id uint) (*model.Message, error) {
message, err := s.repo.FindByAccountAndID(ctx, accountID, id)
if err != nil {
return err
return nil, err
}
if err := s.repo.Delete(ctx, message.ID); err != nil {
return err
message.Content = "This message was deleted"
message.ContentType = "text"
message.ContentAttributes = datatypes.JSON([]byte(`{"deleted":true}`))
if err := s.repo.DB().WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Save(message).Error; err != nil {
return err
}
if err := tx.Where("message_id = ?", message.ID).Delete(&model.Attachment{}).Error; err != nil {
return err
}
return nil
}); err != nil {
return nil, err
}
// Dispatch EventMessageDeleted
s.dispatchMessageEvent(ctx, channel.EventMessageDeleted, message)
s.deleteMessageIndex(ctx, accountID, message.ID)
return nil
return message, nil
}
// UpdateStatus updates the delivery status of a message and dispatches EventMessageStatusUpdated.
@@ -296,6 +363,62 @@ func (s *MessageService) UpdateStatus(ctx context.Context, id uint, status strin
return message, nil
}
func (s *MessageService) ListByConversationFinder(ctx context.Context, conversationID uint, after, before uint, filterInternal bool) ([]model.Message, int64, error) {
return s.repo.FindByConversationFinder(ctx, conversationID, after, before, filterInternal)
}
func validMessageStatus(value string) bool {
switch value {
case "sent", "delivered", "read", "failed":
return true
default:
return false
}
}
func setMessageExternalError(attrs datatypes.JSON, status, externalError string) datatypes.JSON {
obj := map[string]any{}
if len(attrs) > 0 {
_ = json.Unmarshal(attrs, &obj)
}
if status == "failed" && strings.TrimSpace(externalError) != "" {
obj["external_error"] = externalError
} else {
delete(obj, "external_error")
}
bytes, _ := json.Marshal(obj)
return datatypes.JSON(bytes)
}
func attachmentFileType(contentType string) string {
contentType = strings.ToLower(contentType)
switch {
case strings.HasPrefix(contentType, "image/"):
return "image"
case strings.HasPrefix(contentType, "audio/"):
return "audio"
case strings.HasPrefix(contentType, "video/"):
return "video"
default:
return "file"
}
}
func attachmentDataURL(messageID uint, fileName string) string {
fileName = strings.TrimSpace(fileName)
if fileName == "" {
return ""
}
return fmt.Sprintf("/uploads/messages/%d/%s", messageID, fileName)
}
func attachmentThumbURL(contentType string, messageID uint, fileName string) string {
if strings.HasPrefix(strings.ToLower(contentType), "image/") {
return attachmentDataURL(messageID, fileName)
}
return ""
}
// Retry retries a failed message by resetting its delivery status.
func (s *MessageService) Retry(ctx context.Context, accountID, id uint) (*model.Message, error) {
message, err := s.repo.FindByAccountAndID(ctx, accountID, id)
+9 -8
View File
@@ -393,13 +393,14 @@ func TestMessageService_Delete(t *testing.T) {
}
require.NoError(t, db.Create(msg).Error)
// 正常路径:软删除消息
err := svc.Delete(ctx, account.ID, msg.ID)
// 正常路径: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)
// 验证已软删除
_, err = svc.GetByAccountAndID(ctx, account.ID, msg.ID)
assert.Error(t, err)
stored, err := svc.GetByAccountAndID(ctx, account.ID, msg.ID)
assert.NoError(t, err)
assert.JSONEq(t, `{"deleted":true}`, string(stored.ContentAttributes))
// 错误路径:accountID不匹配
msg2 := &model.Message{
@@ -407,11 +408,11 @@ func TestMessageService_Delete(t *testing.T) {
Content: "另一条消息", MessageType: "incoming", ContentType: "text", SenderType: "contact",
}
require.NoError(t, db.Create(msg2).Error)
err = svc.Delete(ctx, 9999, msg2.ID)
_, err = svc.Delete(ctx, 9999, msg2.ID)
assert.Error(t, err)
// 错误路径:ID不存在
err = svc.Delete(ctx, account.ID, 9999)
_, err = svc.Delete(ctx, account.ID, 9999)
assert.Error(t, err)
}
@@ -621,4 +622,4 @@ func TestMessageService_Retry(t *testing.T) {
_, err := svc.Retry(ctx, 9999, msg.ID)
assert.Error(t, err)
})
}
}
@@ -100,7 +100,8 @@ func TestMessageService_SearchIndexHooks(t *testing.T) {
require.NoError(t, err)
_, err = svc.Update(context.Background(), account.ID, message.ID, UpdateMessageRequest{Content: "updated"})
require.NoError(t, err)
require.NoError(t, svc.Delete(context.Background(), account.ID, message.ID))
_, err = svc.Delete(context.Background(), account.ID, message.ID)
require.NoError(t, err)
assert.Equal(t, []string{"message", "message"}, indexer.indexed)
assert.Equal(t, []string{"message"}, indexer.deleted)
+2 -1
View File
@@ -53,6 +53,7 @@ func setupServiceTestDB(t *testing.T) *gorm.DB {
&model.Conversation{},
&model.ConversationParticipant{},
&model.Message{},
&model.Attachment{},
&model.InboxMember{},
&model.Notification{},
&model.NotificationPreference{},
@@ -372,4 +373,4 @@ func createTestMessage(t *testing.T, db *gorm.DB, accountID, inboxID, conversati
func skipIfSQLite(t *testing.T) {
t.Helper()
t.Skip("Skipping: this test requires PostgreSQL (ILIKE / trigram / pgvector etc.)")
}
}