From 74249d5bf55d920d149a826511c15df19f12edad Mon Sep 17 00:00:00 2001 From: Rogee Date: Sun, 7 Jun 2026 14:03:48 +0800 Subject: [PATCH] feat(crm): queue contact bulk actions --- docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md | 13 +- .../handler/api/v1/bulk_action_handler.go | 103 ++++++++++++---- .../api/v1/bulk_action_handler_test.go | 47 ++++++++ .../conversation_maintenance_worker.go | 114 +++++++++++++++++- .../conversation_maintenance_worker_test.go | 83 +++++++++++++ 5 files changed, 327 insertions(+), 33 deletions(-) diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index df21aab0..c845d01f 100644 --- a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md +++ b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md @@ -49,11 +49,11 @@ Hermes task landing checklist: ## Current Baseline -- Current tracking checkpoint: 2026-06-07 P3.102 enterprise account limits and billing-route parity, prepared as `feat(enterprise): align account limits API`. -- Latest implementation checkpoint: this checkpoint, prepared as `feat(enterprise): align account limits API`. -- Latest documentation/tooling checkpoint: this tracker update records the enterprise account API route expansion, Chatwoot account-limit payload shape, deletion toggle side effects, and regenerated route parity artifacts. +- Current tracking checkpoint: 2026-06-07 P3.103 contact bulk-action parity, prepared as `feat(crm): queue contact bulk actions`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(crm): queue contact bulk actions`. +- Latest documentation/tooling checkpoint: this tracker update records Chatwoot contact bulk action enqueueing, label add/remove side effects, delete side effects, and raw empty `200 OK` account bulk-action responses. - Plan landing status: complete for the current known Hermes plans and user-confirmed scope. Future work should update this file directly instead of opening a parallel tracker. -- Worktree status at this implementation checkpoint: the reused dashboard enterprise account client routes from `reference/chatwoot/app/javascript/dashboard/api/enterprise/account.js` and `routes.rb:523-527` are registered under `/enterprise/api/v1/accounts/:account_id/*`. `limits` returns Chatwoot-shaped `{ id, limits }` usage data, `toggle_deletion` writes/removes Chatwoot deletion custom attributes, `subscription` records the local customer-creation guard, and checkout/top-up endpoints return explicit billing-provider errors instead of missing routes. This retains P6.1 webhook placeholder burn-down, P3.101 WhatsApp call route-parameter parity, P3.100 dashboard app route-parameter parity, and prior checkpoints. Live API/browser/enterprise smoke still needs the full PostgreSQL/Redis/Meilisearch/GoChat/Vite/Chrome stack. +- Worktree status at this implementation checkpoint: reused dashboard contact bulk actions from `ContactsIndex.vue` and `api/bulkActions.js` now enqueue Chatwoot-shaped contact jobs on the `medium` queue, normalize lower-case `type`, allow empty `ids` like the reference controller/job path, return empty `200 OK`, add/remove contact labels account-scoped, and soft-delete only selected contacts in the current account. This retains P3.102 enterprise account route parity, P6.1 webhook placeholder burn-down, P3.101 WhatsApp call route-parameter parity, P3.100 dashboard app route-parameter parity, and prior checkpoints. Live API/browser/enterprise smoke still needs the full PostgreSQL/Redis/Meilisearch/GoChat/Vite/Chrome stack. - Next executable implementation checkpoint: continue Phase 2/3 drift audit for the next reused-frontend mismatch, or run B12 live smoke when the full PostgreSQL/Redis/Meilisearch/GoChat/Vite/Chrome stack is available. Re-run Phase 6 placeholder audit after future route/smoke changes. - `go test ./...` passes when run outside the restricted socket sandbox for the latest implementation baseline; the latest docs/tooling checkpoint verified `scripts/parity_frontend_smoke.sh --check` with workspace-local temp/cache dirs after `/tmp` was full. - Route dump succeeds with `972` registered routes after enterprise account route tracking. @@ -156,6 +156,7 @@ This table is the shortest authoritative handoff view. If an older lower section | Priority | Workstream | Current state | Next checkpoint | Commit close rule | | --- | --- | --- | --- | --- | +| 0 | P3.103 contact bulk actions | Implemented for reused CRM contact index actions: `POST /api/v1/accounts/:account_id/bulk_actions` now normalizes Chatwoot `type`, accepts Contact payloads without requiring `action_name`, enqueues durable `contact:bulk_action` jobs on the `medium` queue, returns empty `200 OK`, applies label add/remove account-scoped through `contact_labels`/`tags`, soft-deletes only current-account selected contacts, and treats unknown contact operations as no-op success like `Contacts::BulkActionService`. | Keep in Review; reopen from B12 contacts smoke or fresh reference evidence for exact Pundit authorization, label serializer, deleted-association cleanup, or notification side effects beyond the inspected controller/service/frontend contract. | Focused BulkActionHandler tests passed; focused conversation maintenance contact bulk-action worker tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. | | 0 | P3.102 enterprise account limits and billing routes | Implemented for reused enterprise account frontend calls: `GET/POST /enterprise/api/v1/accounts/:account_id/{limits,checkout,subscription,toggle_deletion,topup_checkout}` are now registered and tracked from `routes.rb:523-527`; `limits` returns Chatwoot-shaped usage data for agents, Captain documents/responses, and default-plan conversation/non-web-inbox counts; `toggle_deletion` mutates `marked_for_deletion_at` and `marked_for_deletion_reason`; `subscription` persists the `is_creating_customer` guard when no Stripe customer exists; checkout/top-up return explicit billing-provider errors instead of missing routes. | Keep in Review; reopen from B12 enterprise account smoke or fresh reference evidence for actual Stripe session creation, cloud-env gating, plan-config defaults, or account deletion notification/cancellation jobs beyond the local persisted boundary. | Focused EnterpriseAccountHandler tests passed; route dump/parity artifacts regenerated to `TOTAL: 972` and `435 exact, 0 method-compatible, 9 parameter-compatible, 0 missing out of 444`; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. | | 0 | P6.1 webhook placeholder fallback burn-down | Implemented for Phase 6 placeholder cleanup: `chatwootParityStub` and the unused webhook placeholder helper are removed from product code. Public webhook nil-handler guards now return explicit `503 { error: "webhook provider unavailable", message: "webhook handler is not configured" }` responses instead of `501 not implemented` placeholder bodies, while wired provider handlers still own real Telegram, WhatsApp, TikTok, LINE, Twilio, Twitter, Instagram, and Shopify behavior. | Keep in Review; reopen from placeholder audit or webhook smoke if a frontend/provider-reachable route returns placeholder/not-implemented content or if a provider handler is missing from normal bootstrap. | Focused router tests cover route boot and nil-handler fallback body/status; placeholder audit refreshed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. | | 0 | P3.101 WhatsApp call route parameter parity | Implemented for reused WhatsApp call controls: account-level show/accept/reject/terminate/upload-recording routes now register Chatwoot's `:id` parameter name from `routes.rb:237-242`, while the handler still accepts legacy `:call_id` mounts. Route parity improves from `425 exact / 14 parameter-compatible` to `430 exact / 9 parameter-compatible` with no missing critical routes. AgentCapacity nested route names stay parameter-compatible in the main Gin router because Gin cannot register `/:id` policy members and `/:agent_capacity_policy_id` nested siblings under the same prefix without a wildcard conflict; the nested handlers now tolerate both parameter-name families. | Keep in Review; reopen from B12 WhatsApp calling smoke or fresh reference evidence for route/action payload drift beyond the inspected enterprise controller/frontend call API contract. | Focused WhatsAppCallHandler, AgentCapacityHandler, and router tests passed; route dump/parity artifacts regenerated; full `go test ./...` passed outside the restricted socket sandbox after retrying the known `internal/worker` SQLite in-memory flake; `git diff --check` passed. | @@ -391,6 +392,7 @@ This ledger records the committed parity checkpoints that future slices should b | Commit | Scope | Verification summary | Follow-up state | | --- | --- | --- | --- | +| `feat(crm): queue contact bulk actions` | Advances P3.103/P5.12 with Chatwoot contact bulk-action parity. GoChat now accepts reused frontend Contact bulk payloads for label add/remove and delete, normalizes `type` like the Rails controller, enqueues `contact:bulk_action` on the `medium` queue when a WorkerPool is configured, returns empty `200 OK`, and replays account-scoped label/delete side effects through durable workers. | Focused BulkActionHandler tests and conversation maintenance worker contact bulk-action tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. | Move P3.103 to Review; continue Phase 2/3 drift audit, Phase 6 placeholder audit, or B12 live smoke. | | `fix(routes): align dashboard app ids` | Advances P3.100 with exact Chatwoot dashboard app member route parameter parity. GoChat now registers `GET/PATCH/PUT/DELETE /api/v1/accounts/:account_id/dashboard_apps/:id`, keeps handler compatibility with legacy local `:dashboard_app_id`, and regenerates route parity artifacts. | Focused DashboardAppHandler and router tests passed; route dump/parity regenerated to `TOTAL: 967` and `425 exact, 0 method-compatible, 14 parameter-compatible, 0 missing out of 439`; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. | Move P3.100 to Review; continue reducing remaining parameter-compatible rows or run B12 live smoke. | | `feat(conversations): align destroy job` | Advances P3.99 with Chatwoot conversation destroy parity. GoChat now returns empty `200 OK` for `DELETE /conversations/:conversation_id`, wires a durable low-priority `conversation:delete_object` job into the WorkerPool, and lets that job perform the existing soft-delete/event/search cleanup while retaining a synchronous fallback without workers. | Focused ConversationService delete/job tests and ConversationHandler delete response tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. | Move P3.99 to Review; continue Phase 2/3 drift audit, Phase 6 placeholder audit, or B12 live smoke. | | `feat(conversations): align transcript delivery` | Advances P3.98 with Chatwoot conversation transcript delivery parity. GoChat now adds the Chatwoot `accounts.limits` field, uses account email transcript availability/rate helpers, maps dashboard transcript disabled/rate-limited responses to raw `402`/empty `429`, delivers transcript emails through the fakeable SMTP boundary, increments outbound-email counters, and filters transcript content to public incoming/outgoing chat messages. Automation `send_email_transcript` now no-ops when disabled and stops recipient fan-out at the configured rate limit. | Focused Account model, ConversationService, ConversationHandler, and ActionService transcript tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. | Move P3.98 to Review; continue Phase 2/3 drift audit, Phase 6 placeholder audit, or B12 live smoke. | @@ -2123,7 +2125,7 @@ Tracking table: | P5.9 | Queue inbound provider webhook processing where Chatwoot defers work. | `webhooks/*_events_job.rb`, `webhooks/*_delivery_job.rb` | webhook handlers, provider services | Provider HTTP ack behavior remains Chatwoot-compatible while persistence/dispatch runs through retryable jobs with signature validation preserved. | Review by `feat(webhook): queue incoming messages` | | P5.10 | Queue outbound message delivery and delivery-status updates. | `send_reply_job.rb`, provider delivery/status jobs | message send/channel services, delivery status handler | Outgoing message creation and provider delivery are separated; retries update message/delivery status exactly once. | Review by `feat(messages): queue send replies` and `feat(messages): queue delivery statuses` | | P5.11 | Queue Captain document sync, crawl, response building, embeddings, and Copilot responses. | Captain document/crawl/response/embedding/Copilot/conversation jobs | `internal/service/captain_document_service.go`, `internal/service/copilot_service.go`, `internal/service/captain_conversation_service.go`, Captain/Copilot services | Existing fakeable disabled/failure gates run under durable jobs; document statuses, Copilot message persistence, and Captain conversation replies survive worker restart. | Review by `feat(captain): queue document syncs`, `feat(captain): queue document crawl jobs`, `feat(captain): queue response embedding jobs`, and `feat(captain): queue copilot response jobs` | -| P5.12 | Queue conversation maintenance jobs. | `trigger_scheduled_items_job.rb`, `campaigns/trigger_oneoff_campaign_job.rb`, `conversations/resolution_job.rb`, `reopen_snoozed_conversations_job.rb`, `update_message_status_job.rb`, `bulk_actions_job.rb` | conversation service/handlers | Auto-resolution, snooze reopen, status updates, and bulk actions are scheduled/retryable with idempotent tests. | Review by `feat(conversations): queue maintenance jobs`, `feat(conversations): queue message status updates`, and `feat(conversations): queue bulk actions` | +| P5.12 | Queue conversation and contact maintenance jobs. | `trigger_scheduled_items_job.rb`, `campaigns/trigger_oneoff_campaign_job.rb`, `conversations/resolution_job.rb`, `reopen_snoozed_conversations_job.rb`, `update_message_status_job.rb`, `bulk_actions_job.rb`, `contacts/bulk_action_job.rb` | conversation/contact service handlers | Auto-resolution, snooze reopen, status updates, conversation bulk actions, and contact label/delete bulk actions are scheduled/retryable with account-scoped tests. | Review by `feat(conversations): queue maintenance jobs`, `feat(conversations): queue message status updates`, `feat(conversations): queue bulk actions`, and `feat(crm): queue contact bulk actions` | | P5.13 | Replace placeholder analytics/report builders that need background aggregation. | reporting jobs/services and report controllers | `internal/service/analytics_service.go`, `internal/service/analytics_query_helpers.go`, `internal/service/reporting_rollup_worker.go`, reporting services | P5.13a derives live/bot/conversation summary/matrix/distribution/outgoing-count report values from persisted rows; P5.13b adds lazy cached rollup freshness, durable `reporting:rollup_day` jobs, and `/reports` metric timeseries index parity. | Review by `feat(reports): derive analytics aggregates` and `feat(reports): add analytics timeseries rollups`; remaining report work should be created as drift slices when frontend/reference exposes additional metrics | P5.1 current checkpoint: @@ -2788,3 +2790,4 @@ Verification milestone gates: - 2026-06-07: P3.101 WhatsApp call route-parameter checkpoint prepared as `fix(routes): align whatsapp call ids`; audited Chatwoot `routes.rb:237-242` and the reused account WhatsApp call route family. GoChat now registers account-level WhatsApp call show/action routes with Chatwoot's `:id` parameter name, preserves legacy `:call_id` handler compatibility, and makes AgentCapacity nested handlers accept both Chatwoot `:agent_capacity_policy_id`/`:id` and local Gin-compatible `:id`/`:user_id`/`:limit_id` parameter names. The main router keeps AgentCapacity nested route names parameter-compatible because Gin rejects wildcard-name changes below the existing policy `/:id` member route. Focused WhatsAppCallHandler, AgentCapacityHandler, and router tests passed; route dump/parity regenerated to `TOTAL: 967` and `430 exact, 0 method-compatible, 9 parameter-compatible, 0 missing out of 439`; full `go test ./...` passed outside the restricted socket sandbox after retrying the known `internal/worker` SQLite in-memory flake; `git diff --check` passed. - 2026-06-07: P6.1 webhook placeholder fallback checkpoint prepared as `fix(webhooks): replace parity stubs`; refreshed the Phase 6 placeholder audit and burned down the remaining `chatwootParityStub` nil-handler fallbacks in `internal/router/router.go`. Public webhook routes now return explicit `503 webhook provider unavailable` JSON when a provider handler is not configured instead of `501 not implemented` placeholder bodies, and the unused placeholder helper is removed. Focused router tests cover boot and nil-handler fallback behavior; placeholder audit shows no `chatwootParityStub` product-code matches; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. - 2026-06-07: P3.102 enterprise account limits checkpoint prepared as `feat(enterprise): align account limits API`; audited Chatwoot enterprise `AccountsController#limits/#toggle_deletion/#subscription/#checkout/#topup_checkout`, `BillingHelper`, `Enterprise::Account::PlanUsageAndLimits`, reused dashboard `api/enterprise/account.js`, and `routes.rb:523-527`. GoChat now registers the enterprise account route family under `/enterprise/api/v1/accounts/:account_id`, returns Chatwoot-shaped account limit payloads, persists scheduled deletion custom attributes, records subscription customer-creation guards, and exposes explicit local billing-provider errors for checkout/top-up paths. Focused EnterpriseAccountHandler tests passed; route dump/parity regenerated to `TOTAL: 972` and `435 exact, 0 method-compatible, 9 parameter-compatible, 0 missing out of 444`; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. +- 2026-06-07: P3.103 contact bulk-action checkpoint prepared as `feat(crm): queue contact bulk actions`; audited Chatwoot `BulkActionsController`, `Contacts::BulkActionJob`, `Contacts::BulkActionService`, reused dashboard `api/bulkActions.js`, and `ContactsIndex.vue` label/delete callers. GoChat account bulk actions now accept Contact payloads for label add/remove and delete, normalize lower-case type values, enqueue durable `contact:bulk_action` jobs on the `medium` queue, return empty `200 OK`, replay account-scoped contact label add/remove through tags/contact_labels, soft-delete only selected current-account contacts, and preserve no-op success for unknown contact bulk payloads. Focused BulkActionHandler and conversation maintenance worker contact bulk-action tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. diff --git a/internal/handler/api/v1/bulk_action_handler.go b/internal/handler/api/v1/bulk_action_handler.go index 7cc1c815..b36ecbbf 100644 --- a/internal/handler/api/v1/bulk_action_handler.go +++ b/internal/handler/api/v1/bulk_action_handler.go @@ -2,6 +2,7 @@ package v1 import ( "net/http" + "strings" "github.com/gin-gonic/gin" @@ -32,9 +33,9 @@ func (h *BulkActionHandler) WithWorkerPool(wp *worker.WorkerPool) *BulkActionHan // BulkActionRequest is the DTO for generic bulk actions. // Reference: Chatwoot bulk_actions_controller#create — params[:type], params[:action_name], params[:ids] type BulkActionRequest struct { - Type string `json:"type" binding:"required"` // "Conversation" or "Contact" - ActionName string `json:"action_name,omitempty"` // legacy local action names - IDs []uint `json:"ids" binding:"required,min=1"` // Chatwoot sends conversation display IDs + Type string `json:"type"` // "Conversation" or "Contact" + ActionName string `json:"action_name,omitempty"` // legacy local action names + IDs []uint `json:"ids"` // Chatwoot sends conversation display IDs Fields service.ConversationBulkActionFields `json:"fields,omitempty"` Labels service.ConversationBulkActionLabels `json:"labels,omitempty"` AssigneeID *uint `json:"assignee_id,omitempty"` // legacy local assign field @@ -58,6 +59,7 @@ func (h *BulkActionHandler) Create(c *gin.Context) { return } + req.Type = normalizeBulkActionType(req.Type) switch req.Type { case "Conversation": h.handleConversationBulk(c, accountID, req) @@ -141,31 +143,80 @@ func (h *BulkActionHandler) handleConversationBulk(c *gin.Context, accountID uin // handleContactBulk processes bulk actions on contacts. // Reference: Chatwoot only supports "delete" and label operations for contacts in bulk_actions. func (h *BulkActionHandler) handleContactBulk(c *gin.Context, accountID uint, req BulkActionRequest) { - successCount := 0 - failCount := 0 - - for _, contactID := range req.IDs { - var svcErr error - - switch req.ActionName { - case "delete": - svcErr = h.contactSvc.Delete(c.Request.Context(), accountID, contactID) - default: - // Chatwoot: other contact actions (labels) require separate async jobs - // For now, return error for unsupported actions - failCount++ - continue + if h.worker != nil { + params := service.ContactBulkActionParams{ + Type: req.Type, + ActionName: req.ActionName, + IDs: req.IDs, + Labels: req.Labels, } - - if svcErr != nil { - failCount++ - } else { - successCount++ + if _, err := service.EnqueueContactBulkAction(c.Request.Context(), h.worker, accountID, getUserID(c), params); err != nil { + response.AbortWithStatusError(c, http.StatusInternalServerError, response.ErrInternal, "failed to enqueue bulk action") + return } + c.Status(http.StatusOK) + return } - response.OK(c, gin.H{ - "success_count": successCount, - "fail_count": failCount, - }) + if err := h.performContactBulkSync(c, accountID, req); err != nil { + response.AbortWithStatusError(c, http.StatusInternalServerError, response.ErrInternal, "failed to process bulk action") + return + } + c.Status(http.StatusOK) +} + +func (h *BulkActionHandler) performContactBulkSync(c *gin.Context, accountID uint, req BulkActionRequest) error { + if h.contactSvc == nil { + return nil + } + switch { + case req.ActionName == "delete": + for _, contactID := range req.IDs { + if err := h.contactSvc.Delete(c.Request.Context(), accountID, contactID); err != nil { + return err + } + } + case len(req.Labels.Add) > 0: + for _, contactID := range req.IDs { + current, err := h.contactSvc.GetLabels(c.Request.Context(), accountID, contactID) + if err != nil { + return err + } + if _, err := h.contactSvc.UpdateLabels(c.Request.Context(), accountID, contactID, append(current, req.Labels.Add...)); err != nil { + return err + } + } + case len(req.Labels.Remove) > 0: + remove := map[string]struct{}{} + for _, label := range req.Labels.Remove { + remove[strings.TrimSpace(label)] = struct{}{} + } + for _, contactID := range req.IDs { + current, err := h.contactSvc.GetLabels(c.Request.Context(), accountID, contactID) + if err != nil { + return err + } + kept := current[:0] + for _, label := range current { + if _, ok := remove[label]; !ok { + kept = append(kept, label) + } + } + if _, err := h.contactSvc.UpdateLabels(c.Request.Context(), accountID, contactID, kept); err != nil { + return err + } + } + } + return nil +} + +func normalizeBulkActionType(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case "conversation": + return "Conversation" + case "contact": + return "Contact" + default: + return strings.TrimSpace(value) + } } diff --git a/internal/handler/api/v1/bulk_action_handler_test.go b/internal/handler/api/v1/bulk_action_handler_test.go index 2698228b..57438e9d 100644 --- a/internal/handler/api/v1/bulk_action_handler_test.go +++ b/internal/handler/api/v1/bulk_action_handler_test.go @@ -47,3 +47,50 @@ func TestBulkActionHandler_ConversationEnqueuesChatwootPayload(t *testing.T) { require.Contains(t, string(job.Payload), `"status":"resolved"`) require.Contains(t, string(job.Payload), `"add":["vip"]`) } + +func TestBulkActionHandler_ContactEnqueuesChatwootPayload(t *testing.T) { + gin.SetMode(gin.TestMode) + db, err := gorm.Open(sqlite.Open("file:bulk-action-contact-handler?mode=memory&cache=private"), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.AutoMigrate(&model.BackgroundJob{})) + sqlDB, _ := db.DB() + t.Cleanup(func() { _ = sqlDB.Close() }) + + wp := worker.NewWorkerPool(db) + handler := NewBulkActionHandler(nil, nil).WithWorkerPool(wp) + router := gin.New() + router.POST("/api/v1/accounts/:account_id/bulk_actions", handler.Create) + + body := []byte(`{"type":"contact","ids":[11,12],"labels":{"add":["vip"],"remove":["old"]}}`) + request := httptest.NewRequest(http.MethodPost, "/api/v1/accounts/7/bulk_actions", bytes.NewReader(body)) + request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-User-ID", "42") + response := httptest.NewRecorder() + router.ServeHTTP(response, request) + + require.Equal(t, http.StatusOK, response.Code) + require.Empty(t, response.Body.String()) + + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ? AND queue = ?", service.TaskTypeContactBulkAction, "medium").First(&job).Error) + require.Contains(t, string(job.Payload), `"account_id":7`) + require.Contains(t, string(job.Payload), `"user_id":42`) + require.Contains(t, string(job.Payload), `"type":"Contact"`) + require.Contains(t, string(job.Payload), `"ids":[11,12]`) + require.Contains(t, string(job.Payload), `"add":["vip"]`) +} + +func TestBulkActionHandler_InvalidTypeMatchesChatwootPayload(t *testing.T) { + gin.SetMode(gin.TestMode) + handler := NewBulkActionHandler(nil, nil) + router := gin.New() + router.POST("/api/v1/accounts/:account_id/bulk_actions", handler.Create) + + request := httptest.NewRequest(http.MethodPost, "/api/v1/accounts/7/bulk_actions", bytes.NewReader([]byte(`{"type":"Ticket"}`))) + request.Header.Set("Content-Type", "application/json") + response := httptest.NewRecorder() + router.ServeHTTP(response, request) + + require.Equal(t, http.StatusUnprocessableEntity, response.Code) + require.JSONEq(t, `{"success":false}`, response.Body.String()) +} diff --git a/internal/service/conversation_maintenance_worker.go b/internal/service/conversation_maintenance_worker.go index 27d71ecb..6d1a038d 100644 --- a/internal/service/conversation_maintenance_worker.go +++ b/internal/service/conversation_maintenance_worker.go @@ -22,6 +22,7 @@ const ( TaskTypeConversationResolutionForAccount = "conversation:resolution" TaskTypeConversationUpdateMessageStatus = "conversation:update_message_status" TaskTypeConversationBulkAction = "conversation:bulk_action" + TaskTypeContactBulkAction = "contact:bulk_action" ) const ( @@ -64,12 +65,25 @@ type ConversationBulkActionLabels struct { Remove []string `json:"remove,omitempty"` } +type ContactBulkActionParams struct { + Type string `json:"type"` + ActionName string `json:"action_name,omitempty"` + IDs []uint `json:"ids"` + Labels ConversationBulkActionLabels `json:"labels,omitempty"` +} + type conversationBulkActionJob struct { AccountID uint `json:"account_id"` UserID uint `json:"user_id,omitempty"` Params ConversationBulkActionParams `json:"params"` } +type contactBulkActionJob struct { + AccountID uint `json:"account_id"` + UserID uint `json:"user_id,omitempty"` + Params ContactBulkActionParams `json:"params"` +} + var conversationMaintenanceRegistrations sync.Map // RegisterConversationMaintenanceJobs wires Chatwoot scheduled maintenance jobs @@ -94,6 +108,7 @@ func registerConversationMaintenanceJobsWithNow(wp *worker.WorkerPool, db *gorm. wp.Register(TaskTypeConversationResolutionForAccount, runner.performResolutionForAccount) wp.Register(TaskTypeConversationUpdateMessageStatus, runner.performUpdateMessageStatus) wp.Register(TaskTypeConversationBulkAction, runner.performConversationBulkAction) + wp.Register(TaskTypeContactBulkAction, runner.performContactBulkAction) } func EnqueueScheduledItemsTrigger(ctx context.Context, wp *worker.WorkerPool, scheduledAt time.Time) (*model.BackgroundJob, error) { @@ -127,7 +142,7 @@ func EnqueueConversationBulkAction(ctx context.Context, wp *worker.WorkerPool, a if wp == nil { return nil, nil } - if accountID == 0 || len(params.IDs) == 0 { + if accountID == 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}, @@ -136,6 +151,19 @@ func EnqueueConversationBulkAction(ctx context.Context, wp *worker.WorkerPool, a ) } +func EnqueueContactBulkAction(ctx context.Context, wp *worker.WorkerPool, accountID, userID uint, params ContactBulkActionParams) (*model.BackgroundJob, error) { + if wp == nil { + return nil, nil + } + if accountID == 0 { + return nil, fmt.Errorf("invalid contact bulk action payload: account_id=%d ids=%v", accountID, params.IDs) + } + return wp.Enqueue(ctx, TaskTypeContactBulkAction, contactBulkActionJob{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) @@ -308,9 +336,12 @@ func (r *conversationMaintenanceRunner) performConversationBulkAction(ctx contex 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 { + if payload.AccountID == 0 { return fmt.Errorf("invalid conversation bulk action job payload: %#v", payload) } + if len(payload.Params.IDs) == 0 { + return nil + } if payload.Params.Type != "" && payload.Params.Type != "Conversation" { return nil } @@ -359,6 +390,85 @@ func (r *conversationMaintenanceRunner) performConversationBulkAction(ctx contex return nil } +func (r *conversationMaintenanceRunner) performContactBulkAction(ctx context.Context, job *model.BackgroundJob) error { + var payload contactBulkActionJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal contact bulk action job: %w", err) + } + if payload.AccountID == 0 { + return fmt.Errorf("invalid contact bulk action job payload: %#v", payload) + } + if payload.Params.Type != "" && payload.Params.Type != "Contact" { + return nil + } + if len(payload.Params.IDs) == 0 { + return nil + } + + switch { + case payload.Params.ActionName == "delete": + return r.db.WithContext(ctx). + Where("account_id = ? AND id IN ?", payload.AccountID, payload.Params.IDs). + Delete(&model.Contact{}).Error + case len(payload.Params.Labels.Add) > 0: + return bulkAddContactLabels(ctx, r.db, payload.AccountID, payload.Params.IDs, payload.Params.Labels.Add) + case len(payload.Params.Labels.Remove) > 0: + return bulkRemoveContactLabels(ctx, r.db, payload.AccountID, payload.Params.IDs, payload.Params.Labels.Remove) + default: + return nil + } +} + +func bulkAddContactLabels(ctx context.Context, db *gorm.DB, accountID uint, contactIDs []uint, labels []string) error { + labels = normalizeContactServiceLabels(labels) + if len(contactIDs) == 0 || len(labels) == 0 { + return nil + } + var scopedContactIDs []uint + if err := db.WithContext(ctx).Model(&model.Contact{}). + Where("account_id = ? AND id IN ?", accountID, contactIDs). + Pluck("id", &scopedContactIDs).Error; err != nil { + return err + } + if len(scopedContactIDs) == 0 { + return nil + } + return db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + for _, label := range labels { + tag := model.Tag{AccountID: accountID, Name: label} + if err := tx.Where("account_id = ? AND name = ?", accountID, label).FirstOrCreate(&tag).Error; err != nil { + return err + } + for _, contactID := range scopedContactIDs { + contactLabel := model.ContactLabel{AccountID: accountID, ContactID: contactID, TagID: tag.ID} + if err := tx.Where("account_id = ? AND contact_id = ? AND tag_id = ?", accountID, contactID, tag.ID).FirstOrCreate(&contactLabel).Error; err != nil { + return err + } + } + } + return nil + }) +} + +func bulkRemoveContactLabels(ctx context.Context, db *gorm.DB, accountID uint, contactIDs []uint, labels []string) error { + labels = normalizeContactServiceLabels(labels) + if len(contactIDs) == 0 || len(labels) == 0 { + return nil + } + var tagIDs []uint + if err := db.WithContext(ctx).Model(&model.Tag{}). + Where("account_id = ? AND name IN ?", accountID, labels). + Pluck("id", &tagIDs).Error; err != nil { + return err + } + if len(tagIDs) == 0 { + return nil + } + return db.WithContext(ctx). + Where("account_id = ? AND contact_id IN ? AND tag_id IN ?", accountID, contactIDs, tagIDs). + Delete(&model.ContactLabel{}).Error +} + func parseBulkActionTime(value string) (time.Time, bool) { if value == "" { return time.Time{}, false diff --git a/internal/service/conversation_maintenance_worker_test.go b/internal/service/conversation_maintenance_worker_test.go index e7ba4e20..805018bd 100644 --- a/internal/service/conversation_maintenance_worker_test.go +++ b/internal/service/conversation_maintenance_worker_test.go @@ -318,6 +318,75 @@ func TestConversationMaintenanceJobsConversationBulkAction(t *testing.T) { } } +func TestConversationMaintenanceJobsContactBulkAction(t *testing.T) { + now := time.Date(2026, 6, 6, 0, 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) + contactA := createTestContact(t, db, account.ID) + contactB := createTestContact(t, db, account.ID) + otherContact := createTestContact(t, db, otherAccount.ID) + oldTag := model.Tag{AccountID: account.ID, Name: "old"} + if err := db.Create(&oldTag).Error; err != nil { + t.Fatalf("create old tag: %v", err) + } + if err := db.Create(&model.ContactLabel{AccountID: account.ID, ContactID: contactA.ID, TagID: oldTag.ID}).Error; err != nil { + t.Fatalf("label contact A: %v", err) + } + if err := db.Create(&model.ContactLabel{AccountID: account.ID, ContactID: contactB.ID, TagID: oldTag.ID}).Error; err != nil { + t.Fatalf("label contact B: %v", err) + } + + _, err := EnqueueContactBulkAction(context.Background(), wp, account.ID, 42, ContactBulkActionParams{ + Type: "Contact", + IDs: []uint{contactA.ID, contactB.ID, otherContact.ID}, + Labels: ConversationBulkActionLabels{Add: []string{"vip", "vip"}}, + }) + if err != nil { + t.Fatalf("enqueue contact label add: %v", err) + } + assertJobCount(t, db, TaskTypeContactBulkAction, 1) + processRequiredJob(t, wp, "contact bulk add labels") + assertContactHasLabel(t, db, account.ID, contactA.ID, "vip", true) + assertContactHasLabel(t, db, account.ID, contactB.ID, "vip", true) + assertContactHasLabel(t, db, otherAccount.ID, otherContact.ID, "vip", false) + + _, err = EnqueueContactBulkAction(context.Background(), wp, account.ID, 42, ContactBulkActionParams{ + Type: "Contact", + IDs: []uint{contactA.ID, contactB.ID}, + Labels: ConversationBulkActionLabels{Remove: []string{"old"}}, + }) + if err != nil { + t.Fatalf("enqueue contact label remove: %v", err) + } + processRequiredJob(t, wp, "contact bulk remove labels") + assertContactHasLabel(t, db, account.ID, contactA.ID, "old", false) + assertContactHasLabel(t, db, account.ID, contactB.ID, "old", false) + assertContactHasLabel(t, db, account.ID, contactA.ID, "vip", true) + + _, err = EnqueueContactBulkAction(context.Background(), wp, account.ID, 42, ContactBulkActionParams{ + Type: "Contact", + ActionName: "delete", + IDs: []uint{contactA.ID, otherContact.ID}, + }) + if err != nil { + t.Fatalf("enqueue contact delete: %v", err) + } + processRequiredJob(t, wp, "contact bulk delete") + + var deleted model.Contact + if err := db.First(&deleted, contactA.ID).Error; err == nil { + t.Fatalf("expected account contact to be soft deleted") + } + var untouched model.Contact + if err := db.First(&untouched, otherContact.ID).Error; err != nil { + t.Fatalf("other account contact should remain visible: %v", err) + } +} + func createTestOneoffCampaign(t *testing.T, db *gorm.DB, accountID, inboxID, contactID uint, scheduledAt time.Time) *campaign.Campaign { t.Helper() c := &campaign.Campaign{ @@ -358,6 +427,20 @@ func createConversationMaintenanceMessage(t *testing.T, db *gorm.DB, accountID, return message } +func assertContactHasLabel(t *testing.T, db *gorm.DB, accountID, contactID uint, label string, want bool) { + t.Helper() + var count int64 + if err := db.Table("contact_labels"). + Joins("JOIN tags ON tags.id = contact_labels.tag_id"). + Where("contact_labels.account_id = ? AND contact_labels.contact_id = ? AND tags.name = ?", accountID, contactID, label). + Count(&count).Error; err != nil { + t.Fatalf("count contact label %s: %v", label, err) + } + if got := count > 0; got != want { + t.Fatalf("contact %d label %q presence = %v, want %v", contactID, label, got, want) + } +} + func assertMessageStatus(t *testing.T, db *gorm.DB, messageID uint, want string) { t.Helper() var message model.Message