feat(automation): queue macro and csat jobs

docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md

@@ -17,9 +17,9 @@ Build GoChat as a Go backend that can directly reuse the frontend from `referenc
 ## Current Baseline
 
 - Plan freeze checkpoint: 2026-06-05 docs-only tracker landing after `13cb750 feat(captain): align task payload persistence`.
-- Latest implementation checkpoint: this checkpoint, prepared as `feat(automation): queue external action deliveries`.
+- Latest implementation checkpoint: this checkpoint, prepared as `feat(automation): queue macro and csat jobs`.
 - Latest documentation checkpoint before this freeze: `3263ed9 docs: land copilot task execution plan`; this document now carries the active follow-up plan directly.
-- Worktree status at this implementation checkpoint: B11.1a aligns Captain assistant CRUD/tools/inbox bindings; B11.1b aligns Captain scenarios and custom tools; B11.1c aligns Captain documents, assistant responses, bulk actions, and custom-tool test payloads; B11.2 aligns Copilot thread/message create/list/get/delete payloads, account/user scoping, and no-LLM fallback persistence; B11.3a aligns Captain preferences show/update payloads and account-level model/feature storage; B11.3b aligns Captain playground request/response payloads, account scoping, v2 history handling, and no-LLM fallback; B11.3c adds the fakeable Captain document sync backend gate with disabled, failed, and fake-success states; B11.3d aligns Captain task request/response payloads, no-provider disabled states, follow-up context, suggestion persistence, and Copilot message tool-call key validation; B11.3e aligns Captain stream DTOs/disabled SSE fallbacks and Copilot push-event payload shapes; B12.1 adds the reusable GoChat server/seed entrypoint plus a Meilisearch-first reused Chatwoot frontend smoke harness and report; B12.2a adds API smoke assertions for auth/profile, inbox, conversation/messages, contact/company, widget config/message, and public CSAT; B12.2b adds a zero-dependency Chrome DevTools browser smoke that loads the reused Chatwoot login and dashboard entrypoints through Vite and checks browser auth/dashboard API requests; B12.3a adds enterprise API smoke assertions for SLA reports/download, CSAT reports/download, automation/macros, audit/custom roles, capacity, Captain, and Copilot; B12.3b adds reused-frontend enterprise browser route navigation for SLA, CSAT, automation, macros, audit logs, custom roles, capacity, Captain, and Copilot request coverage; P5.1 adds the PostgreSQL-backed durable `background_jobs` model/migration plus WorkerPool enqueue, schedule, retry/backoff, dead-letter, idempotency, stale-lock recovery, and focused tests; P5.2 wires `channel.Dispatcher` and `dispatch.EventDispatcher` async paths into durable event jobs with worker replay tests; P5.4 queues automation webhook and email transcript side effects as durable jobs while preserving fakeable delivery boundaries. Next active implementation slice is P5.5/P5.6 delayed automation, macro, and CSAT job integration.
+- Worktree status at this implementation checkpoint: B11.1a aligns Captain assistant CRUD/tools/inbox bindings; B11.1b aligns Captain scenarios and custom tools; B11.1c aligns Captain documents, assistant responses, bulk actions, and custom-tool test payloads; B11.2 aligns Copilot thread/message create/list/get/delete payloads, account/user scoping, and no-LLM fallback persistence; B11.3a aligns Captain preferences show/update payloads and account-level model/feature storage; B11.3b aligns Captain playground request/response payloads, account scoping, v2 history handling, and no-LLM fallback; B11.3c adds the fakeable Captain document sync backend gate with disabled, failed, and fake-success states; B11.3d aligns Captain task request/response payloads, no-provider disabled states, follow-up context, suggestion persistence, and Copilot message tool-call key validation; B11.3e aligns Captain stream DTOs/disabled SSE fallbacks and Copilot push-event payload shapes; B12.1 adds the reusable GoChat server/seed entrypoint plus a Meilisearch-first reused Chatwoot frontend smoke harness and report; B12.2a adds API smoke assertions for auth/profile, inbox, conversation/messages, contact/company, widget config/message, and public CSAT; B12.2b adds a zero-dependency Chrome DevTools browser smoke that loads the reused Chatwoot login and dashboard entrypoints through Vite and checks browser auth/dashboard API requests; B12.3a adds enterprise API smoke assertions for SLA reports/download, CSAT reports/download, automation/macros, audit/custom roles, capacity, Captain, and Copilot; B12.3b adds reused-frontend enterprise browser route navigation for SLA, CSAT, automation, macros, audit logs, custom roles, capacity, Captain, and Copilot request coverage; P5.1 adds the PostgreSQL-backed durable `background_jobs` model/migration plus WorkerPool enqueue, schedule, retry/backoff, dead-letter, idempotency, stale-lock recovery, and focused tests; P5.2 wires `channel.Dispatcher` and `dispatch.EventDispatcher` async paths into durable event jobs with worker replay tests; P5.4 queues automation webhook and email transcript side effects as durable jobs while preserving fakeable delivery boundaries; P5.5 queues Chatwoot-style macro execute fan-out through durable `automation:macro_execution` jobs; P5.6 queues resolve-triggered CSAT survey sends through idempotent durable jobs. Next active implementation slice is delayed automation actions, CSAT WhatsApp/Twilio templates, and Meilisearch/SLA job fan-out.
 - `go test ./...` passes.
 - Route dump succeeds with `TOTAL: 830` after adding the Chatwoot-compatible applied-SLA index route.
 - Route parity artifacts now exist under `docs/parity/` and are generated by `cmd/route_parity`.
@@ -78,7 +78,7 @@ Open work after the current checkpoint:
 | Phase 2 | Route and controller parity audit | Doing | Ruby/Bundler unavailable, so Chatwoot route extraction currently uses static `routes.rb` fallback |
 | Phase 3 | Data and serializer parity | Doing | JSON fixture coverage is partial and still endpoint-family based |
 | Phase 4 | Enterprise feature completion | Doing | B7, B8, B9, B10, and B11 are in Review; B12 reused frontend smoke is the next broad verification gate |
-| Phase 5 | Background jobs and integrations | Doing | P5.1/P5.2/P5.4 durable worker, event dispatch, and automation external-delivery cores are in Review; delayed automation/search/SLA/Captain job integration remains open |
+| Phase 5 | Background jobs and integrations | Doing | P5.1/P5.2/P5.4/P5.5/P5.6 durable worker, event dispatch, automation delivery, macro, and CSAT cores are in Review; delayed automation/search/SLA/Captain job integration remains open |
 | Phase 6 | Core placeholder burn-down | Doing | account/contact/conversation/message/inbox placeholder groups remain broad |
 | Phase 7 | Verification harness | Review | B12.1 boot/readiness, B12.2a API assertions, B12.2b browser smoke harness, B12.3a enterprise API assertions, and B12.3b enterprise browser route navigation exist; optional live Meilisearch/full-browser runs remain environment-dependent |
 
@@ -101,6 +101,7 @@ This ledger records the committed parity checkpoints that future slices should b
 
 | Commit | Scope | Verification summary | Follow-up state |
 | --- | --- | --- | --- |
+| `feat(automation): queue macro and csat jobs` | Advances P5.5/P5.6 with durable macro fan-out and resolve-triggered CSAT sends. `MacroService.ExecuteForDisplayIDs` now enqueues `automation:macro_execution` jobs when a WorkerPool is configured, resolving display IDs at perform time like Chatwoot `MacrosExecutionJob`; `CsatSurveyListener` now enqueues idempotent `csat:survey_send` jobs for resolved conversations while keeping message-updated response building inline. | `go test ./internal/automation ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.5/P5.6 section. | Move macro fan-out and generic CSAT send to Review; continue delayed automation actions and WhatsApp/Twilio CSAT template jobs. |
 | `feat(automation): queue external action deliveries` | Advances P5.4 with durable automation external side effects. `send_webhook_event` and `send_email_transcript` now enqueue `automation:webhook_delivery` and `automation:transcript_delivery` jobs when a WorkerPool is configured; automation rules and macros can carry the WorkerPool through `ActionService`, and app bootstrap starts the durable worker plus dispatcher event jobs. | `go test ./internal/automation ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.4 section. | Move P5.4 to Review; continue P5.5 delayed automation/macro fan-out and P5.6 CSAT channel-template jobs. |
 | `feat(dispatch): queue async events durably` | Advances P5.2 with durable async event dispatch. `channel.Dispatcher.DispatchAsync` now enqueues `event:dispatch_async` jobs when a WorkerPool is configured, and `dispatch.EventDispatcher` now enqueues per-listener `event:listener_dispatch` jobs for async listener replay while preserving immediate sync listener behavior. | `go test ./internal/channel ./internal/dispatch ./internal/worker -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.2 section. | Move P5.2 to Review; continue P5.4/P5.5 automation webhook/transcript and CSAT durable job integration. |
 | `feat(worker): add durable background jobs` | Advances P5.1 with a persistent `background_jobs` model and migration plus `internal/worker.WorkerPool` enqueue, scheduled claim, handler registry, retry/backoff, dead-letter, idempotency-key reuse, queue filtering, PostgreSQL `SKIP LOCKED` claim support, stale lock recovery, and graceful start/stop loop. The model is registered in app auto-migration and test DB defaults. | `go test ./internal/worker -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.1 section. | Move P5.1 to Review; continue P5.2 dispatcher async routing and P5.4/P5.5 automation/CSAT job integration on top of the durable core. |
@@ -1609,8 +1610,8 @@ Tracking table:
 | P5.2 | Route async dispatcher events through durable jobs. | `event_dispatcher_job.rb`, Chatwoot async dispatcher listeners | `internal/dispatch/dispatcher.go`, `internal/channel/dispatcher.go` | Heavy listeners can enqueue durable jobs without changing sync listener behavior; tests cover sync vs async routing and replay. | Review by `feat(dispatch): queue async events durably` |
 | P5.3 | Move Meilisearch indexing and reindex fan-out into retryable jobs. | Meilisearch plan plus Chatwoot callbacks/jobs that index searchable records | search services, contact/company/conversation indexing hooks | Create/update/delete indexing survives handler success, retries on Meilisearch failure, and optional live Meilisearch gate remains green. | Todo |
 | P5.4 | Queue automation webhook and transcript delivery. | `webhook_job.rb`, automation action execution services | `internal/automation/action_delivery.go`, `internal/automation/action_service.go` | Existing timeout/retry fakeable delivery is invoked by durable jobs; logs preserve attempt metadata and idempotency. | Review by `feat(automation): queue external action deliveries` |
-| P5.5 | Queue delayed automation actions and macro execution. | `trigger_scheduled_items_job.rb`, `macros_execution_job.rb` | automation rule listener, macro service | Delayed actions execute after schedule time, macro execute supports multi-conversation job fan-out, and repeated workers do not duplicate side effects. | Todo |
-| P5.6 | Queue CSAT survey sends and channel-specific templates. | CSAT listener/services, WhatsApp/Twilio template services/jobs | `internal/automation/csat_survey_listener.go`, `internal/csat/listener.go`, channel send services | Resolve-triggered CSAT send is durable; WhatsApp/Twilio template delivery and failure states are fakeable and observable. | Todo |
+| P5.5 | Queue delayed automation actions and macro execution. | `trigger_scheduled_items_job.rb`, `macros_execution_job.rb` | automation rule listener, macro service | Delayed actions execute after schedule time, macro execute supports multi-conversation job fan-out, and repeated workers do not duplicate side effects. | Doing: macro fan-out Review by `feat(automation): queue macro and csat jobs`; delayed actions Todo |
+| P5.6 | Queue CSAT survey sends and channel-specific templates. | CSAT listener/services, WhatsApp/Twilio template services/jobs | `internal/automation/csat_survey_listener.go`, `internal/csat/listener.go`, channel send services | Resolve-triggered CSAT send is durable; WhatsApp/Twilio template delivery and failure states are fakeable and observable. | Doing: generic survey send Review by `feat(automation): queue macro and csat jobs`; WhatsApp/Twilio templates Todo |
 | P5.7 | Queue SLA account-wide scans and applied-SLA processing. | `enterprise/app/jobs/sla/trigger_slas_for_accounts_job.rb`, `process_account_applied_slas_job.rb`, `process_applied_sla_job.rb` | `internal/service/applied_sla_service.go`, SLA event services | Scheduler enqueues account scans and per-SLA processing; threshold transitions remain idempotent and notification fan-out is retryable. | Todo |
 | P5.8 | Queue contact export artifact generation and completion notification/email. | `account/contacts_export_job.rb` | contact export service and mailer boundary | Export API returns immediately, artifact generation is durable, completion notification/email is fakeable, and download remains stable after restart. | Todo |
 | 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. | Todo |
@@ -1667,6 +1668,22 @@ env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache
 git diff --check
 ```
 
+P5.5/P5.6 current checkpoint:
+
+- `MacroService.ExecuteForDisplayIDs` now follows Chatwoot `MacrosExecutionJob` when a WorkerPool is configured: request handlers can return `200 OK` after enqueue, the worker later resolves account-scoped conversation display IDs, and macro actions execute per conversation under the existing macro/action service boundary.
+- Macro job replay logs execution rows after each conversation and does not retry solely because the audit-style execution record failed, avoiding duplicate conversation side effects from log-only failures.
+- `CsatSurveyListener` now queues resolve-triggered CSAT sends through `csat:survey_send` jobs with an idempotency key per conversation. Duplicate resolved events reuse the same queued/completed job, and the worker still relies on `SendSurveyForConversationID`'s existing one-survey-message guard.
+- Message-updated CSAT response building remains inline because it is part of the submitted widget/public API update path, not an external delivery side effect.
+- Remaining P5.5/P5.6 work: delayed automation scheduled-item execution and WhatsApp/Twilio CSAT template creation/status/send jobs.
+
+P5.5/P5.6 verification:
+
+```bash
+env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/automation ./internal/worker ./internal/app -count=1
+env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./...
+git diff --check
+```
+
 ## Phase 6: Core Product Placeholder Burn-down
 
 Status: doing.
@@ -1861,3 +1878,4 @@ Verification milestone gates:
 - 2026-06-05: P5.1 durable worker checkpoint prepared as `feat(worker): add durable background jobs`; the `background_jobs` table/model and WorkerPool core now cover enqueueing, scheduled jobs, queue filtering, idempotency keys, PostgreSQL `SKIP LOCKED` claims, handler execution, retry/backoff, dead-letter state, stale-lock recovery, and graceful start/stop. Focused worker tests cover idempotent enqueue, completion, retries, schedule/queue gating, and restart pickup. Next slice is P5.2/P5.4 integration of dispatcher and automation jobs onto the durable core.
 - 2026-06-05: P5.2 durable event dispatch checkpoint prepared as `feat(dispatch): queue async events durably`; `channel.Dispatcher.DispatchAsync` now enqueues event replay jobs when a WorkerPool is configured, and `dispatch.EventDispatcher` enqueues per-listener async jobs while preserving inline sync listeners. Focused channel/dispatch/worker tests prove durable enqueue and worker replay. Next slice is P5.4/P5.5 automation webhook/transcript and CSAT job integration.
 - 2026-06-05: P5.4 durable automation delivery checkpoint prepared as `feat(automation): queue external action deliveries`; automation and macro `send_webhook_event`/`send_email_transcript` actions now enqueue durable jobs when a WorkerPool is configured, worker handlers invoke the existing fakeable HTTP/SMTP delivery boundaries, and app bootstrap starts the shared durable worker for dispatcher and automation queues. Focused automation/worker/app tests passed. Next slice is P5.5/P5.6 delayed automation, macro fan-out, and CSAT job integration.
+- 2026-06-05: P5.5/P5.6 durable macro and CSAT checkpoint prepared as `feat(automation): queue macro and csat jobs`; macro execute now enqueues `automation:macro_execution` jobs for display-ID fan-out when a WorkerPool is configured, and resolved-conversation CSAT sends now enqueue idempotent `csat:survey_send` jobs. Focused automation/worker/app tests passed. Next slice is delayed automation scheduled items and WhatsApp/Twilio CSAT template delivery.

internal/app/bootstrap.go

@@ -394,7 +394,7 @@ func Bootstrap(env string) (*App, error) {
 	// M6: Register automation rule listener on the dispatcher for event-triggered automation
 	automation.RegisterAutomationRuleListenerWithWorker(channelDispatcher, &dbProvider{db: db}, workerPool)
 	// M6: Register CSAT survey listener on the dispatcher for conversation.resolved events
-	channelDispatcher.Register(automation.NewCsatSurveyListener(&dbProvider{db: db}))
+	channelDispatcher.Register(automation.NewCsatSurveyListenerWithWorker(&dbProvider{db: db}, workerPool))
 
 	// Create Facebook webhook handler (Gin HTTP handler for FB/IG webhook endpoints)
 	facebookWebhookHandler := webhook.NewFacebookWebhookHandler(fbProvider, igProvider, db, channelDispatcher)

internal/automation/csat_survey_listener.go

@@ -2,17 +2,23 @@ package automation
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 
 	"github.com/gochat/gochat/internal/channel"
 	"github.com/gochat/gochat/internal/model"
+	"github.com/gochat/gochat/internal/worker"
 	applogger "github.com/gochat/gochat/pkg/logger"
 	"gorm.io/gorm"
 )
 
+const TaskTypeCsatSurveySend = "csat:survey_send"
+
 // CsatSurveyListener sends CSAT surveys when conversations are resolved.
 // Reference: Chatwoot AutomationRuleListener triggers SendCSATSurvey action on conversation.resolved
 type CsatSurveyListener struct {
-	db DBProvider
+	db     DBProvider
+	worker *worker.WorkerPool
 }
 
 // NewCsatSurveyListener creates a new CsatSurveyListener.
@@ -20,6 +26,19 @@ func NewCsatSurveyListener(db DBProvider) *CsatSurveyListener {
 	return &CsatSurveyListener{db: db}
 }
 
+func NewCsatSurveyListenerWithWorker(db DBProvider, wp *worker.WorkerPool) *CsatSurveyListener {
+	listener := NewCsatSurveyListener(db)
+	listener.SetWorkerPool(wp)
+	return listener
+}
+
+func (l *CsatSurveyListener) SetWorkerPool(wp *worker.WorkerPool) {
+	l.worker = wp
+	if wp != nil {
+		wp.Register(TaskTypeCsatSurveySend, l.performSurveySendJob)
+	}
+}
+
 // Name returns the listener name for registration.
 func (l *CsatSurveyListener) Name() string {
 	return "csat_survey_listener"
@@ -39,7 +58,14 @@ func (l *CsatSurveyListener) OnEvent(ctx context.Context, event *channel.Channel
 		applogger.L().Warnf("csat_survey_listener: missing conversation_id in event %s", event.Type)
 		return nil
 	}
+	if l.worker != nil {
+		_, err := l.worker.Enqueue(ctx, TaskTypeCsatSurveySend, csatSurveySendJob{ConversationID: conversationID}, worker.WithQueue("automation"), worker.WithMaxAttempts(3), worker.WithIdempotencyKey(fmt.Sprintf("csat-survey:%d", conversationID)))
+		return err
+	}
+	return l.sendSurvey(ctx, conversationID)
+}
 
+func (l *CsatSurveyListener) sendSurvey(ctx context.Context, conversationID uint) error {
 	message, err := NewCsatSurveyService(l.db).SendSurveyForConversationID(ctx, conversationID)
 	if err != nil {
 		applogger.L().Errorf("csat_survey_listener: failed to send CSAT survey for conversation %d: %v", conversationID, err)
@@ -52,6 +78,21 @@ func (l *CsatSurveyListener) OnEvent(ctx context.Context, event *channel.Channel
 	return nil
 }
 
+type csatSurveySendJob struct {
+	ConversationID uint `json:"conversation_id"`
+}
+
+func (l *CsatSurveyListener) performSurveySendJob(ctx context.Context, job *model.BackgroundJob) error {
+	var payload csatSurveySendJob
+	if err := json.Unmarshal(job.Payload, &payload); err != nil {
+		return fmt.Errorf("unmarshal csat survey send job: %w", err)
+	}
+	if payload.ConversationID == 0 {
+		return fmt.Errorf("csat survey send job missing conversation_id")
+	}
+	return l.sendSurvey(ctx, payload.ConversationID)
+}
+
 func (l *CsatSurveyListener) onMessageUpdated(ctx context.Context, event *channel.ChannelEvent) error {
 	messageID, ok := eventUint(event, "message_id")
 	if !ok {

internal/automation/csat_survey_listener_test.go

@@ -4,9 +4,11 @@ import (
 	"context"
 	"encoding/json"
 	"testing"
+	"time"
 
 	"github.com/gochat/gochat/internal/channel"
 	"github.com/gochat/gochat/internal/model"
+	"github.com/gochat/gochat/internal/worker"
 	"gorm.io/datatypes"
 	"gorm.io/gorm"
 )
@@ -77,6 +79,48 @@ func TestCsatSurveyListener_ResolvedConversationSkipsDisabledInbox(t *testing.T)
 	}
 }
 
+func TestCsatSurveyListener_ResolvedConversationQueuesDurableSurveySend(t *testing.T) {
+	dbProvider := setupAutomationTestDBProvider(t)
+	db := dbProvider.DB()
+	accountID, _ := seedTestAccount(db, t)
+	conversation := seedCsatListenerConversation(t, db, accountID, true)
+	wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 13, 30, 0, 0, time.UTC) }))
+	listener := NewCsatSurveyListenerWithWorker(dbProvider, wp)
+	event := &channel.ChannelEvent{Type: channel.EventConversationResolved, ConversationID: conversation.ID}
+
+	if err := listener.OnEvent(context.Background(), event); err != nil {
+		t.Fatalf("queue csat survey: %v", err)
+	}
+	if err := listener.OnEvent(context.Background(), event); err != nil {
+		t.Fatalf("queue duplicate csat survey: %v", err)
+	}
+	var messageCount int64
+	if err := db.Model(&model.Message{}).Where("conversation_id = ? AND content_type = ?", conversation.ID, "input_csat").Count(&messageCount).Error; err != nil {
+		t.Fatalf("count csat messages before worker: %v", err)
+	}
+	if messageCount != 0 {
+		t.Fatalf("csat survey should not send synchronously, got %d messages", messageCount)
+	}
+	var jobCount int64
+	if err := db.Model(&model.BackgroundJob{}).Where("job_type = ? AND status = ?", TaskTypeCsatSurveySend, model.BackgroundJobStatusQueued).Count(&jobCount).Error; err != nil {
+		t.Fatalf("count csat jobs: %v", err)
+	}
+	if jobCount != 1 {
+		t.Fatalf("expected one idempotent csat job, got %d", jobCount)
+	}
+
+	processed, err := wp.ProcessOne(context.Background())
+	if err != nil || !processed {
+		t.Fatalf("process csat job: processed=%v err=%v", processed, err)
+	}
+	if err := db.Model(&model.Message{}).Where("conversation_id = ? AND content_type = ?", conversation.ID, "input_csat").Count(&messageCount).Error; err != nil {
+		t.Fatalf("count csat messages after worker: %v", err)
+	}
+	if messageCount != 1 {
+		t.Fatalf("expected one durable csat message, got %d", messageCount)
+	}
+}
+
 func TestCsatSurveyListener_MessageUpdatedBuildsResponse(t *testing.T) {
 	dbProvider := setupAutomationTestDBProvider(t)
 	db := dbProvider.DB()

internal/automation/macro_service.go

@@ -19,6 +19,8 @@ type MacroService struct {
 	worker *worker.WorkerPool
 }
 
+const TaskTypeMacroExecution = "automation:macro_execution"
+
 // NewMacroService creates a new MacroService.
 func NewMacroService(db DBProvider) *MacroService {
 	return &MacroService{db: db}
@@ -33,6 +35,7 @@ func NewMacroServiceWithWorker(db DBProvider, wp *worker.WorkerPool) *MacroServi
 func (s *MacroService) SetWorkerPool(wp *worker.WorkerPool) {
 	s.worker = wp
 	RegisterActionDeliveryJobs(wp, s.db)
+	RegisterMacroExecutionJobs(wp, s.db)
 }
 
 // GetByID retrieves a macro by ID.
@@ -206,6 +209,30 @@ func (s *MacroService) Execute(ctx context.Context, accountID uint, conversation
 // ExecuteForDisplayIDs runs a macro against account-scoped conversation display IDs.
 // Reference: Chatwoot MacrosExecutionJob resolves account.conversations by display_id.
 func (s *MacroService) ExecuteForDisplayIDs(ctx context.Context, accountID uint, macroID uint, displayIDs []uint, userID uint) error {
+	if s.worker != nil {
+		return s.enqueueExecuteForDisplayIDs(ctx, accountID, macroID, displayIDs, userID)
+	}
+	return s.executeForDisplayIDsNow(ctx, accountID, macroID, displayIDs, userID)
+}
+
+func (s *MacroService) enqueueExecuteForDisplayIDs(ctx context.Context, accountID uint, macroID uint, displayIDs []uint, userID uint) error {
+	if len(displayIDs) == 0 {
+		return nil
+	}
+	_, err := s.GetByIDForAccount(ctx, accountID, macroID)
+	if err != nil {
+		return err
+	}
+	_, err = s.worker.Enqueue(ctx, TaskTypeMacroExecution, macroExecutionJob{
+		AccountID:       accountID,
+		MacroID:         macroID,
+		ConversationIDs: displayIDs,
+		UserID:          userID,
+	}, worker.WithQueue("medium"), worker.WithMaxAttempts(3))
+	return err
+}
+
+func (s *MacroService) executeForDisplayIDsNow(ctx context.Context, accountID uint, macroID uint, displayIDs []uint, userID uint) error {
 	macro, err := s.GetByIDForAccount(ctx, accountID, macroID)
 	if err != nil {
 		return err
@@ -239,7 +266,7 @@ func (s *MacroService) ExecuteForDisplayIDs(ctx context.Context, accountID uint,
 			}
 		}
 		if err := s.recordExecution(ctx, macroID, conversation.ID, userID); err != nil && err != gorm.ErrRecordNotFound {
-			return err
+			applogger.L().Warnf("failed to record macro %d execution for conversation %d: %v", macroID, conversation.ID, err)
 		}
 	}
 	return nil

internal/automation/macro_service_test.go

@@ -3,8 +3,10 @@ package automation
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/gochat/gochat/internal/model"
+	"github.com/gochat/gochat/internal/worker"
 )
 
 func TestMacroService_Create(t *testing.T) {
@@ -13,15 +15,15 @@ func TestMacroService_Create(t *testing.T) {
 	svc := NewMacroService(dbProvider)
 
 	macro := &Macro{
-		AccountID:    accountID,
-		Name:         "Close and label",
+		AccountID: accountID,
+		Name:      "Close and label",
 		Actions: Actions{
 			{ActionName: "resolve", ActionParams: map[string]interface{}{}},
 			{ActionName: "add_label", ActionParams: map[string]interface{}{"labels": []string{"closed"}}},
 		},
-		Visibility:   MacroVisibilityGlobal,
-		CreatedByID:  userID,
-		UpdatedByID:  userID,
+		Visibility:  MacroVisibilityGlobal,
+		CreatedByID: userID,
+		UpdatedByID: userID,
 	}
 
 	err := svc.Create(context.Background(), macro)
@@ -39,12 +41,12 @@ func TestMacroService_GetByID(t *testing.T) {
 	svc := NewMacroService(dbProvider)
 
 	macro := &Macro{
-		AccountID:    accountID,
-		Name:         "Test macro",
-		Actions:      Actions{},
-		Visibility:   MacroVisibilityGlobal,
-		CreatedByID:  userID,
-		UpdatedByID:  userID,
+		AccountID:   accountID,
+		Name:        "Test macro",
+		Actions:     Actions{},
+		Visibility:  MacroVisibilityGlobal,
+		CreatedByID: userID,
+		UpdatedByID: userID,
 	}
 	if err := svc.Create(context.Background(), macro); err != nil {
 		t.Fatalf("setup: %v", err)
@@ -190,3 +192,67 @@ func TestMacroService_Delete(t *testing.T) {
 		t.Fatal("expected error getting deleted macro, got nil")
 	}
 }
+
+func TestMacroService_ExecuteForDisplayIDsQueuesDurableMacroJob(t *testing.T) {
+	dbProvider := setupAutomationTestDBProvider(t)
+	db := dbProvider.DB()
+	accountID, userID := seedTestAccount(db, t)
+	inboxID := seedTestInbox(db, t, accountID)
+	contactID := seedTestContact(db, t, accountID)
+	displayID := uint(616)
+	conversationID := seedTestConversationWithDetails(db, t, accountID, inboxID, contactID, "open", "low", "web", 0)
+	if err := db.Model(&model.Conversation{}).Where("id = ?", conversationID).Update("display_id", displayID).Error; err != nil {
+		t.Fatalf("set display id: %v", err)
+	}
+	macro := &Macro{
+		AccountID: accountID,
+		Name:      "durable macro",
+		Actions: Actions{
+			{ActionName: "add_label", ActionParams: map[string]interface{}{"labels": []string{"durable"}}},
+			{ActionName: "change_status", ActionParams: map[string]interface{}{"status": "resolved"}},
+		},
+		Visibility:  MacroVisibilityGlobal,
+		CreatedByID: userID,
+		UpdatedByID: userID,
+	}
+	if err := NewMacroService(dbProvider).Create(context.Background(), macro); err != nil {
+		t.Fatalf("create macro: %v", err)
+	}
+	wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 13, 15, 0, 0, time.UTC) }))
+	svc := NewMacroServiceWithWorker(dbProvider, wp)
+
+	if err := svc.ExecuteForDisplayIDs(context.Background(), accountID, macro.ID, []uint{displayID}, userID); err != nil {
+		t.Fatalf("queue macro execution: %v", err)
+	}
+	var before model.Conversation
+	if err := db.First(&before, conversationID).Error; err != nil {
+		t.Fatalf("load conversation before worker: %v", err)
+	}
+	if before.Status != "open" {
+		t.Fatalf("macro should not run synchronously, status=%s", before.Status)
+	}
+
+	var count int64
+	if err := db.Model(&model.BackgroundJob{}).Where("job_type = ? AND queue = ? AND status = ?", TaskTypeMacroExecution, "medium", model.BackgroundJobStatusQueued).Count(&count).Error; err != nil {
+		t.Fatalf("count macro jobs: %v", err)
+	}
+	if count != 1 {
+		t.Fatalf("expected one queued macro job, got %d", count)
+	}
+	processed, err := wp.ProcessOne(context.Background())
+	if err != nil || !processed {
+		t.Fatalf("process macro job: processed=%v err=%v", processed, err)
+	}
+
+	var after model.Conversation
+	if err := db.First(&after, conversationID).Error; err != nil {
+		t.Fatalf("load conversation after worker: %v", err)
+	}
+	if after.Status != "resolved" {
+		t.Fatalf("expected macro job to resolve conversation, got %s", after.Status)
+	}
+	var label ConversationLabel
+	if err := db.Where("conversation_id = ? AND label = ?", conversationID, "durable").First(&label).Error; err != nil {
+		t.Fatalf("expected macro job to add label: %v", err)
+	}
+}

internal/automation/macro_worker.go

@@ -0,0 +1,46 @@
+package automation
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+
+	"github.com/gochat/gochat/internal/model"
+	"github.com/gochat/gochat/internal/worker"
+)
+
+type macroExecutionJob struct {
+	AccountID       uint   `json:"account_id"`
+	MacroID         uint   `json:"macro_id"`
+	ConversationIDs []uint `json:"conversation_ids"`
+	UserID          uint   `json:"user_id"`
+}
+
+var macroExecutionRegistrations sync.Map
+
+// RegisterMacroExecutionJobs wires Chatwoot's MacrosExecutionJob equivalent.
+// The job resolves account-scoped conversation display IDs at perform time.
+func RegisterMacroExecutionJobs(wp *worker.WorkerPool, db DBProvider) {
+	if wp == nil || db == nil {
+		return
+	}
+	if _, loaded := macroExecutionRegistrations.LoadOrStore(wp, struct{}{}); loaded {
+		return
+	}
+	runner := &macroExecutionJobRunner{db: db, worker: wp}
+	wp.Register(TaskTypeMacroExecution, runner.perform)
+}
+
+type macroExecutionJobRunner struct {
+	db     DBProvider
+	worker *worker.WorkerPool
+}
+
+func (r *macroExecutionJobRunner) perform(ctx context.Context, job *model.BackgroundJob) error {
+	var payload macroExecutionJob
+	if err := json.Unmarshal(job.Payload, &payload); err != nil {
+		return fmt.Errorf("unmarshal macro execution job: %w", err)
+	}
+	return NewMacroServiceWithWorker(r.db, r.worker).executeForDisplayIDsNow(ctx, payload.AccountID, payload.MacroID, payload.ConversationIDs, payload.UserID)
+}