feat(conversations): queue bulk actions

2026-06-05 18:16:56 +08:00
parent 530f11bcd8
commit cda7d0266e
6 changed files with 335 additions and 16 deletions
@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strings"
 	"sync"
 	"time"

@@ -20,6 +21,7 @@ const (
 	TaskTypeConversationResolutionScheduler  = "account:conversations_resolution_scheduler"
 	TaskTypeConversationResolutionForAccount = "conversation:resolution"
 	TaskTypeConversationUpdateMessageStatus  = "conversation:update_message_status"
+	TaskTypeConversationBulkAction           = "conversation:bulk_action"
 )

 const (
@@ -42,6 +44,32 @@ type conversationUpdateMessageStatusJob struct {
 	Status         string `json:"status"`
 }

+type ConversationBulkActionParams struct {
+	Type         string                       `json:"type"`
+	ActionName   string                       `json:"action_name,omitempty"`
+	IDs          []uint                       `json:"ids"`
+	Fields       ConversationBulkActionFields `json:"fields,omitempty"`
+	Labels       ConversationBulkActionLabels `json:"labels,omitempty"`
+	SnoozedUntil string                       `json:"snoozed_until,omitempty"`
+}
+
+type ConversationBulkActionFields struct {
+	Status     *string `json:"status,omitempty"`
+	AssigneeID *uint   `json:"assignee_id,omitempty"`
+	TeamID     *uint   `json:"team_id,omitempty"`
+}
+
+type ConversationBulkActionLabels struct {
+	Add    []string `json:"add,omitempty"`
+	Remove []string `json:"remove,omitempty"`
+}
+
+type conversationBulkActionJob struct {
+	AccountID uint                         `json:"account_id"`
+	UserID    uint                         `json:"user_id,omitempty"`
+	Params    ConversationBulkActionParams `json:"params"`
+}
+
 var conversationMaintenanceRegistrations sync.Map

 // RegisterConversationMaintenanceJobs wires Chatwoot scheduled maintenance jobs
@@ -65,6 +93,7 @@ func registerConversationMaintenanceJobsWithNow(wp *worker.WorkerPool, db *gorm.
 	wp.Register(TaskTypeConversationResolutionScheduler, runner.performResolutionScheduler)
 	wp.Register(TaskTypeConversationResolutionForAccount, runner.performResolutionForAccount)
 	wp.Register(TaskTypeConversationUpdateMessageStatus, runner.performUpdateMessageStatus)
+	wp.Register(TaskTypeConversationBulkAction, runner.performConversationBulkAction)
 }

 func EnqueueScheduledItemsTrigger(ctx context.Context, wp *worker.WorkerPool, scheduledAt time.Time) (*model.BackgroundJob, error) {
@@ -94,6 +123,19 @@ func EnqueueConversationMessageStatusUpdate(ctx context.Context, wp *worker.Work
 	)
 }

+func EnqueueConversationBulkAction(ctx context.Context, wp *worker.WorkerPool, accountID, userID uint, params ConversationBulkActionParams) (*model.BackgroundJob, error) {
+	if wp == nil {
+		return nil, nil
+	}
+	if accountID == 0 || len(params.IDs) == 0 {
+		return nil, fmt.Errorf("invalid conversation bulk action payload: account_id=%d ids=%v", accountID, params.IDs)
+	}
+	return wp.Enqueue(ctx, TaskTypeConversationBulkAction, conversationBulkActionJob{AccountID: accountID, UserID: userID, Params: params},
+		worker.WithQueue("medium"),
+		worker.WithMaxAttempts(3),
+	)
+}
+
 func scheduledItemsIdempotencyKey(scheduledAt time.Time) string {
 	bucket := scheduledAt.UTC().Truncate(scheduledItemsInterval).Unix()
 	return fmt.Sprintf("scheduled:trigger_items:%d", bucket)
@@ -260,3 +302,113 @@ func validConversationMessageStatus(status string) bool {
 		return false
 	}
 }
+
+func (r *conversationMaintenanceRunner) performConversationBulkAction(ctx context.Context, job *model.BackgroundJob) error {
+	var payload conversationBulkActionJob
+	if err := json.Unmarshal(job.Payload, &payload); err != nil {
+		return fmt.Errorf("unmarshal conversation bulk action job: %w", err)
+	}
+	if payload.AccountID == 0 || len(payload.Params.IDs) == 0 {
+		return fmt.Errorf("invalid conversation bulk action job payload: %#v", payload)
+	}
+	if payload.Params.Type != "" && payload.Params.Type != "Conversation" {
+		return nil
+	}
+
+	var conversations []model.Conversation
+	if err := r.db.WithContext(ctx).
+		Where("account_id = ? AND display_id IN ?", payload.AccountID, payload.Params.IDs).
+		Find(&conversations).Error; err != nil {
+		return fmt.Errorf("load bulk action conversations: %w", err)
+	}
+	for i := range conversations {
+		conversation := conversations[i]
+		updates := map[string]any{}
+		if payload.Params.Fields.Status != nil && *payload.Params.Fields.Status != "" {
+			updates["status"] = *payload.Params.Fields.Status
+			now := r.now()
+			switch *payload.Params.Fields.Status {
+			case string(model.ConversationStatusResolved):
+				updates["resolved_at"] = now
+			case string(model.ConversationStatusOpen):
+				updates["resumed_at"] = now
+				updates["snoozed_until"] = nil
+			}
+		}
+		if payload.Params.Fields.AssigneeID != nil {
+			updates["assignee_id"] = payload.Params.Fields.AssigneeID
+		}
+		if payload.Params.Fields.TeamID != nil {
+			updates["team_id"] = payload.Params.Fields.TeamID
+		}
+		if payload.Params.SnoozedUntil != "" {
+			if snoozedUntil, ok := parseBulkActionTime(payload.Params.SnoozedUntil); ok {
+				updates["snoozed_until"] = snoozedUntil.Unix()
+			}
+		}
+		if len(payload.Params.Labels.Add) > 0 || len(payload.Params.Labels.Remove) > 0 {
+			updates["labels"] = mergeConversationLabels(conversation.Labels, payload.Params.Labels.Add, payload.Params.Labels.Remove)
+		}
+		if len(updates) == 0 {
+			continue
+		}
+		if err := r.db.WithContext(ctx).Model(&model.Conversation{}).Where("id = ?", conversation.ID).Updates(updates).Error; err != nil {
+			return fmt.Errorf("bulk update conversation %d: %w", conversation.ID, err)
+		}
+	}
+	return nil
+}
+
+func parseBulkActionTime(value string) (time.Time, bool) {
+	if value == "" {
+		return time.Time{}, false
+	}
+	if ts, err := time.Parse(time.RFC3339, value); err == nil {
+		return ts.UTC(), true
+	}
+	if ts, err := time.Parse("2006-01-02T15:04:05.000Z", value); err == nil {
+		return ts.UTC(), true
+	}
+	return time.Time{}, false
+}
+
+func mergeConversationLabels(existing string, add, remove []string) string {
+	labels := map[string]bool{}
+	order := []string{}
+	for _, label := range splitConversationLabels(existing) {
+		if !labels[label] {
+			labels[label] = true
+			order = append(order, label)
+		}
+	}
+	for _, label := range add {
+		label = strings.TrimSpace(label)
+		if label == "" || labels[label] {
+			continue
+		}
+		labels[label] = true
+		order = append(order, label)
+	}
+	for _, label := range remove {
+		delete(labels, strings.TrimSpace(label))
+	}
+	merged := make([]string, 0, len(order))
+	for _, label := range order {
+		if labels[label] {
+			merged = append(merged, label)
+		}
+	}
+	return strings.Join(merged, ",")
+}
+
+func splitConversationLabels(existing string) []string {
+	parts := strings.Split(existing, ",")
+	labels := make([]string, 0, len(parts))
+	for _, part := range parts {
+		label := strings.TrimSpace(part)
+		if label != "" {
+			labels = append(labels, label)
+		}
+	}
+	return labels
+}
@@ -3,6 +3,7 @@ package service
 import (
 	"context"
 	"fmt"
+	"strings"
 	"testing"
 	"time"

@@ -237,6 +238,86 @@ func TestConversationMaintenanceJobsIgnoreInvalidMessageStatus(t *testing.T) {
 	assertMessageStatus(t, db, message.ID, string(model.MessageStatusSent))
 }

+func TestConversationMaintenanceJobsConversationBulkAction(t *testing.T) {
+	now := time.Date(2026, 6, 5, 23, 0, 0, 0, time.UTC)
+	db := setupServiceTestDB(t)
+	wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now }))
+	registerConversationMaintenanceJobsWithNow(wp, db, func() time.Time { return now })
+
+	account := createTestAccount(t, db)
+	otherAccount := createTestAccount(t, db)
+	inbox := createTestInbox(t, db, account.ID, "web_widget")
+	contact := createTestContact(t, db, account.ID)
+	convA := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
+	convB := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
+	otherConv := createTestConversation(t, db, otherAccount.ID, inbox.ID, contact.ID)
+	displayA := uint(101)
+	displayB := uint(102)
+	sharedOtherDisplay := displayA
+	if err := db.Model(convA).Updates(map[string]any{"display_id": displayA, "labels": "vip,old"}).Error; err != nil {
+		t.Fatalf("set display A: %v", err)
+	}
+	if err := db.Model(convB).Updates(map[string]any{"display_id": displayB, "labels": "old"}).Error; err != nil {
+		t.Fatalf("set display B: %v", err)
+	}
+	if err := db.Model(otherConv).Updates(map[string]any{"display_id": sharedOtherDisplay, "labels": "other"}).Error; err != nil {
+		t.Fatalf("set display other: %v", err)
+	}
+	status := string(model.ConversationStatusSnoozed)
+	teamID := uint(77)
+	assigneeID := uint(88)
+	snoozedUntil := now.Add(time.Hour).Format(time.RFC3339)
+
+	_, err := EnqueueConversationBulkAction(context.Background(), wp, account.ID, 42, ConversationBulkActionParams{
+		Type: "Conversation",
+		IDs:  []uint{displayA, displayB},
+		Fields: ConversationBulkActionFields{
+			Status:     &status,
+			TeamID:     &teamID,
+			AssigneeID: &assigneeID,
+		},
+		Labels:       ConversationBulkActionLabels{Add: []string{"urgent", "vip"}, Remove: []string{"old"}},
+		SnoozedUntil: snoozedUntil,
+	})
+	if err != nil {
+		t.Fatalf("enqueue bulk action: %v", err)
+	}
+	assertJobCount(t, db, TaskTypeConversationBulkAction, 1)
+	var queued model.BackgroundJob
+	if err := db.Where("job_type = ?", TaskTypeConversationBulkAction).First(&queued).Error; err != nil {
+		t.Fatalf("load bulk action job: %v", err)
+	}
+	if queued.Queue != "medium" {
+		t.Fatalf("expected medium queue, got %s", queued.Queue)
+	}
+
+	processRequiredJob(t, wp, "conversation bulk action")
+
+	for _, id := range []uint{convA.ID, convB.ID} {
+		var conversation model.Conversation
+		if err := db.First(&conversation, id).Error; err != nil {
+			t.Fatalf("load conversation %d: %v", id, err)
+		}
+		if conversation.Status != status || conversation.TeamID == nil || *conversation.TeamID != teamID || conversation.AssigneeID == nil || *conversation.AssigneeID != assigneeID {
+			t.Fatalf("conversation %d not bulk updated: status=%s team=%v assignee=%v", id, conversation.Status, conversation.TeamID, conversation.AssigneeID)
+		}
+		if conversation.SnoozedUntil == nil || *conversation.SnoozedUntil != now.Add(time.Hour).Unix() {
+			t.Fatalf("conversation %d snoozed_until not updated: %v", id, conversation.SnoozedUntil)
+		}
+		if strings.Contains(conversation.Labels, "old") || !strings.Contains(conversation.Labels, "urgent") || !strings.Contains(conversation.Labels, "vip") {
+			t.Fatalf("conversation %d labels not merged, got %q", id, conversation.Labels)
+		}
+	}
+
+	var untouched model.Conversation
+	if err := db.First(&untouched, otherConv.ID).Error; err != nil {
+		t.Fatalf("load other conversation: %v", err)
+	}
+	if untouched.Labels != "other" || untouched.Status != string(model.ConversationStatusOpen) {
+		t.Fatalf("other account conversation should not change: status=%s labels=%s", untouched.Status, untouched.Labels)
+	}
+}
+
 func createTestOneoffCampaign(t *testing.T, db *gorm.DB, accountID, inboxID, contactID uint, scheduledAt time.Time) *campaign.Campaign {
 	t.Helper()
 	c := &campaign.Campaign{