feat(automation): queue external action deliveries
This commit is contained in:
@@ -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(dispatch): queue async events durably`.
|
||||
- Latest implementation checkpoint: this checkpoint, prepared as `feat(automation): queue external action deliveries`.
|
||||
- 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. Next active implementation slice is P5.4/P5.5 automation 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. Next active implementation slice is P5.5/P5.6 delayed automation, macro, and CSAT job integration.
|
||||
- `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 durable worker and event dispatch cores are in Review; automation/search/SLA/Captain job integration remains open |
|
||||
| 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 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 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. |
|
||||
| `test(parity): add enterprise browser smoke coverage` | Advances B12.3b with `scripts/parity_frontend_smoke.sh --enterprise-browser-smoke`. The browser driver now supports `--enterprise`, creates route-specific reused Chatwoot dashboard shells, and asserts browser-origin requests for SLA reports, CSAT reports, automation rules, macros, audit logs, custom roles, agent capacity list/edit, Captain settings/assistants, and Copilot thread requests. The smoke seed now enables the Chatwoot enterprise feature flags required by those route guards. | `bash -n scripts/parity_frontend_smoke.sh`; `node --check scripts/parity_frontend_browser_smoke.mjs`; `scripts/parity_frontend_smoke.sh --print`; `scripts/parity_frontend_smoke.sh --check`; `go test ./cmd/gochat -count=1`; full verification recorded in the B12.3b section. Live enterprise browser smoke requires running GoChat and Chatwoot Vite. | Move B12 verification harness to Review; continue Phase 5 job parity and convert any optional live-smoke failures into named slices. |
|
||||
@@ -1607,7 +1608,7 @@ Tracking table:
|
||||
| P5.1 | Implement durable worker core and job model. | `reference/chatwoot/app/jobs/application_job.rb`, `mutex_application_job.rb` | `internal/worker/worker.go`, `internal/model/background_job.go`, `migrations/000026_add_background_jobs.*.sql` | Job table, worker persistence API, enqueue API, worker loop, retry/backoff, scheduled jobs, mutex/idempotency keys, dead-letter state, and restart pickup tests exist. | Review by `feat(worker): add durable background jobs` |
|
||||
| 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. | 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.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 |
|
||||
@@ -1650,6 +1651,22 @@ env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache
|
||||
git diff --check
|
||||
```
|
||||
|
||||
P5.4 current checkpoint:
|
||||
|
||||
- `ActionService` can now be constructed with a `WorkerPool`. When present, `send_webhook_event` enqueues `automation:webhook_delivery` jobs and `send_email_transcript` enqueues one `automation:transcript_delivery` job per parsed recipient, matching Chatwoot's async `WebhookJob`/mailer boundary instead of depending on the request path.
|
||||
- The durable job handlers reuse the existing fakeable HTTP webhook and SMTP transcript delivery interfaces, so timeout/retry behavior remains tested at the delivery boundary while worker attempts, errors, retry schedules, completion, and dead-letter state are visible through `background_jobs`.
|
||||
- `AutomationRuleService`, `AutomationRuleListener`, and `MacroService` can carry the WorkerPool through action execution. `Bootstrap` now creates a durable WorkerPool, registers automation delivery jobs, wires the dispatcher with event jobs, and starts the worker in `App.Run`.
|
||||
- Existing no-worker constructors still preserve the previous synchronous fallback for focused tests and development paths that do not start the durable worker yet.
|
||||
- Remaining Phase 5 work: delayed automation actions, macro fan-out jobs, CSAT channel-specific sends/templates, Meilisearch indexing fan-out, SLA scans, contact export email, provider webhook/offline delivery, Captain/Copilot, and conversation maintenance.
|
||||
|
||||
P5.4 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.
|
||||
@@ -1843,3 +1860,4 @@ Verification milestone gates:
|
||||
- 2026-06-05: B12.3b enterprise browser smoke checkpoint prepared as `test(parity): add enterprise browser smoke coverage`; `scripts/parity_frontend_smoke.sh --enterprise-browser-smoke` now runs enterprise API smoke, verifies Chatwoot Vite, and drives reused Chatwoot enterprise dashboard routes through the Chrome DevTools browser harness. Browser assertions cover SLA, CSAT, automation, macros, audit logs, custom roles, agent capacity, Captain settings/assistants, and Copilot thread request paths. Seed now enables the corresponding enterprise feature flags for route guards. Verification: `bash -n`, `node --check`, `--print`, `--check`, `go test ./cmd/gochat`, full `go test ./...`, and `git diff --check` passed locally. Next slice is Phase 5 job parity.
|
||||
- 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.
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
channelmodel "github.com/gochat/gochat/internal/model/channel"
|
||||
"github.com/gochat/gochat/internal/pubsub"
|
||||
"github.com/gochat/gochat/internal/service"
|
||||
"github.com/gochat/gochat/internal/worker"
|
||||
applogger "github.com/gochat/gochat/pkg/logger"
|
||||
"gorm.io/driver/postgres"
|
||||
"gorm.io/gorm"
|
||||
@@ -31,6 +32,7 @@ type App struct {
|
||||
engine *gin.Engine
|
||||
wsHub *ws.Hub
|
||||
notificationDeliverySvc *service.NotificationDeliveryService
|
||||
workerPool *worker.WorkerPool
|
||||
}
|
||||
|
||||
// New creates and initializes the application.
|
||||
@@ -80,6 +82,13 @@ func New(cfg *config.Config) (*App, error) {
|
||||
|
||||
// Run starts the notification delivery pipeline and the HTTP server.
|
||||
func (a *App) Run() error {
|
||||
if a.workerPool != nil {
|
||||
if err := a.workerPool.Start(); err != nil {
|
||||
return fmt.Errorf("failed to start background worker: %w", err)
|
||||
}
|
||||
applogger.L().Info("Background worker started")
|
||||
}
|
||||
|
||||
// Start notification delivery service (Watermill router) in background
|
||||
if a.notificationDeliverySvc != nil {
|
||||
go func() {
|
||||
|
||||
@@ -39,6 +39,7 @@ import (
|
||||
"github.com/gochat/gochat/internal/search"
|
||||
"github.com/gochat/gochat/internal/security"
|
||||
"github.com/gochat/gochat/internal/service"
|
||||
"github.com/gochat/gochat/internal/worker"
|
||||
wspkg "github.com/gochat/gochat/internal/ws"
|
||||
applogger "github.com/gochat/gochat/pkg/logger"
|
||||
"github.com/spf13/viper"
|
||||
@@ -103,6 +104,9 @@ func Bootstrap(env string) (*App, error) {
|
||||
applogger.L().Info("Database migrations completed successfully")
|
||||
}
|
||||
|
||||
workerPool := worker.NewWorkerPool(db)
|
||||
automation.RegisterActionDeliveryJobs(workerPool, &dbProvider{db: db})
|
||||
|
||||
// Step 5: Connect to Redis (ref: Chatwoot config/cable.yml)
|
||||
rdb, err := NewRedisClient(&cfg.Redis)
|
||||
if err != nil {
|
||||
@@ -319,7 +323,7 @@ func Bootstrap(env string) (*App, error) {
|
||||
notionIntegrationService := service.NewNotionIntegrationService(integrationHookRepo)
|
||||
|
||||
// Create channel dispatcher for event-driven architecture (ref: Chatwoot Dispatcher)
|
||||
channelDispatcher := channel.NewDispatcher()
|
||||
channelDispatcher := channel.NewDispatcher(workerPool)
|
||||
|
||||
// P9: AgentBot rule engine services (declared early so listener can be registered on dispatcher)
|
||||
botRuleService := automation.NewBotRuleService(&dbProvider{db: db})
|
||||
@@ -388,7 +392,7 @@ func Bootstrap(env string) (*App, error) {
|
||||
// P9: Register AgentBot rule listener on the dispatcher
|
||||
channelDispatcher.Register(agentBotRuleListener)
|
||||
// M6: Register automation rule listener on the dispatcher for event-triggered automation
|
||||
automation.RegisterAutomationRuleListener(channelDispatcher, &dbProvider{db: db})
|
||||
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}))
|
||||
|
||||
@@ -604,8 +608,8 @@ func Bootstrap(env string) (*App, error) {
|
||||
portalMemberService := service.NewPortalMemberService(portalMemberRepo)
|
||||
|
||||
// Automation services (P6 — Automation Rules + Macros + CSAT)
|
||||
automationRuleService := automation.NewAutomationRuleService(&dbProvider{db: db})
|
||||
macroService := automation.NewMacroService(&dbProvider{db: db})
|
||||
automationRuleService := automation.NewAutomationRuleServiceWithWorker(&dbProvider{db: db}, workerPool)
|
||||
macroService := automation.NewMacroServiceWithWorker(&dbProvider{db: db}, workerPool)
|
||||
csatSurveyService := automation.NewCsatSurveyService(&dbProvider{db: db})
|
||||
cannedResponseService := canned.NewCannedResponseService(&dbProvider{db: db})
|
||||
|
||||
@@ -879,6 +883,7 @@ func Bootstrap(env string) (*App, error) {
|
||||
engine: engine,
|
||||
wsHub: wsHub,
|
||||
notificationDeliverySvc: notificationDeliverySvc,
|
||||
workerPool: workerPool,
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -17,8 +17,10 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
defaultActionDeliveryAttempts = 3
|
||||
defaultActionDeliveryTimeout = 10 * time.Second
|
||||
defaultActionDeliveryAttempts = 3
|
||||
defaultActionDeliveryTimeout = 10 * time.Second
|
||||
TaskTypeAutomationWebhookDelivery = "automation:webhook_delivery"
|
||||
TaskTypeAutomationTranscriptDelivery = "automation:transcript_delivery"
|
||||
)
|
||||
|
||||
// ActionDeliveryResult is copied into AutomationExecution.action_results so
|
||||
@@ -30,6 +32,7 @@ type ActionDeliveryResult struct {
|
||||
ResponseCode int
|
||||
ResponseBody string
|
||||
Retryable bool
|
||||
Queued bool
|
||||
}
|
||||
|
||||
type AutomationWebhookRequest struct {
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
package automation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"github.com/gochat/gochat/internal/model"
|
||||
"github.com/gochat/gochat/internal/worker"
|
||||
)
|
||||
|
||||
type automationWebhookDeliveryJob struct {
|
||||
AccountID uint `json:"account_id"`
|
||||
ConversationID uint `json:"conversation_id"`
|
||||
EventName string `json:"event_name"`
|
||||
URL string `json:"url"`
|
||||
Payload map[string]interface{} `json:"payload"`
|
||||
}
|
||||
|
||||
type automationTranscriptDeliveryJob struct {
|
||||
AccountID uint `json:"account_id"`
|
||||
ConversationID uint `json:"conversation_id"`
|
||||
Recipient string `json:"recipient"`
|
||||
}
|
||||
|
||||
var actionDeliveryRegistrations sync.Map
|
||||
|
||||
// RegisterActionDeliveryJobs wires Chatwoot-style automation external side
|
||||
// effects into the durable worker. Callers may invoke this repeatedly; worker
|
||||
// handlers are idempotently registered per WorkerPool instance.
|
||||
func RegisterActionDeliveryJobs(wp *worker.WorkerPool, db DBProvider) {
|
||||
if wp == nil || db == nil {
|
||||
return
|
||||
}
|
||||
if _, loaded := actionDeliveryRegistrations.LoadOrStore(wp, struct{}{}); loaded {
|
||||
return
|
||||
}
|
||||
runner := &actionDeliveryJobRunner{
|
||||
db: db,
|
||||
webhookDeliverer: defaultWebhookDelivererFactory(),
|
||||
transcriptDeliverer: defaultTranscriptDelivererFactory(),
|
||||
}
|
||||
wp.Register(TaskTypeAutomationWebhookDelivery, runner.performWebhookDelivery)
|
||||
wp.Register(TaskTypeAutomationTranscriptDelivery, runner.performTranscriptDelivery)
|
||||
}
|
||||
|
||||
type actionDeliveryJobRunner struct {
|
||||
db DBProvider
|
||||
webhookDeliverer AutomationWebhookDeliverer
|
||||
transcriptDeliverer AutomationTranscriptDeliverer
|
||||
}
|
||||
|
||||
func (r *actionDeliveryJobRunner) performWebhookDelivery(ctx context.Context, job *model.BackgroundJob) error {
|
||||
var payload automationWebhookDeliveryJob
|
||||
if err := json.Unmarshal(job.Payload, &payload); err != nil {
|
||||
return fmt.Errorf("unmarshal automation webhook job: %w", err)
|
||||
}
|
||||
_, err := r.webhookDeliverer.DeliverWebhook(ctx, AutomationWebhookRequest{
|
||||
AccountID: payload.AccountID,
|
||||
ConversationID: payload.ConversationID,
|
||||
EventName: payload.EventName,
|
||||
URL: payload.URL,
|
||||
Payload: payload.Payload,
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
func (r *actionDeliveryJobRunner) performTranscriptDelivery(ctx context.Context, job *model.BackgroundJob) error {
|
||||
var payload automationTranscriptDeliveryJob
|
||||
if err := json.Unmarshal(job.Payload, &payload); err != nil {
|
||||
return fmt.Errorf("unmarshal automation transcript job: %w", err)
|
||||
}
|
||||
subject, body, err := NewActionService(r.db).buildTranscriptEmail(ctx, payload.AccountID, payload.ConversationID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = r.transcriptDeliverer.DeliverTranscript(ctx, AutomationTranscriptRequest{
|
||||
AccountID: payload.AccountID,
|
||||
ConversationID: payload.ConversationID,
|
||||
Recipient: payload.Recipient,
|
||||
Subject: subject,
|
||||
Body: body,
|
||||
})
|
||||
return err
|
||||
}
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/gochat/gochat/internal/model"
|
||||
"github.com/gochat/gochat/internal/worker"
|
||||
applogger "github.com/gochat/gochat/pkg/logger"
|
||||
)
|
||||
|
||||
@@ -29,6 +30,7 @@ type ActionService struct {
|
||||
db DBProvider
|
||||
webhookDeliverer AutomationWebhookDeliverer
|
||||
transcriptDeliverer AutomationTranscriptDeliverer
|
||||
worker *worker.WorkerPool
|
||||
}
|
||||
|
||||
var defaultWebhookDelivererFactory = func() AutomationWebhookDeliverer {
|
||||
@@ -48,6 +50,17 @@ func NewActionService(db DBProvider) *ActionService {
|
||||
}
|
||||
}
|
||||
|
||||
func NewActionServiceWithWorker(db DBProvider, wp *worker.WorkerPool) *ActionService {
|
||||
s := NewActionService(db)
|
||||
s.SetWorkerPool(wp)
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *ActionService) SetWorkerPool(wp *worker.WorkerPool) {
|
||||
s.worker = wp
|
||||
RegisterActionDeliveryJobs(wp, s.db)
|
||||
}
|
||||
|
||||
func setAutomationActionDeliverersForTest(webhook AutomationWebhookDeliverer, transcript AutomationTranscriptDeliverer) func() {
|
||||
originalWebhookFactory := defaultWebhookDelivererFactory
|
||||
originalTranscriptFactory := defaultTranscriptDelivererFactory
|
||||
@@ -310,6 +323,19 @@ func (s *ActionService) handleSendWebhookEvent(ctx context.Context, accountID, c
|
||||
if err != nil {
|
||||
return ActionDeliveryResult{DeliveryType: "webhook", Target: url}, err
|
||||
}
|
||||
if s.worker != nil {
|
||||
_, err := s.worker.Enqueue(ctx, TaskTypeAutomationWebhookDelivery, automationWebhookDeliveryJob{
|
||||
AccountID: accountID,
|
||||
ConversationID: conversationID,
|
||||
EventName: eventName,
|
||||
URL: url,
|
||||
Payload: payload,
|
||||
}, worker.WithQueue("automation"), worker.WithMaxAttempts(defaultActionDeliveryAttempts))
|
||||
if err != nil {
|
||||
return ActionDeliveryResult{DeliveryType: "webhook", Target: url}, err
|
||||
}
|
||||
return ActionDeliveryResult{DeliveryType: "webhook", Target: url, ResponseBody: "queued", Queued: true}, nil
|
||||
}
|
||||
return s.webhookDeliverer.DeliverWebhook(ctx, AutomationWebhookRequest{
|
||||
AccountID: accountID,
|
||||
ConversationID: conversationID,
|
||||
@@ -391,6 +417,20 @@ func (s *ActionService) handleSendEmailTranscript(ctx context.Context, accountID
|
||||
return ActionDeliveryResult{DeliveryType: "email_transcript"}, fmt.Errorf("send_email_transcript action requires 'email' param")
|
||||
}
|
||||
|
||||
if s.worker != nil {
|
||||
for _, recipient := range recipients {
|
||||
_, err := s.worker.Enqueue(ctx, TaskTypeAutomationTranscriptDelivery, automationTranscriptDeliveryJob{
|
||||
AccountID: accountID,
|
||||
ConversationID: conversationID,
|
||||
Recipient: recipient,
|
||||
}, worker.WithQueue("automation"), worker.WithMaxAttempts(defaultActionDeliveryAttempts))
|
||||
if err != nil {
|
||||
return ActionDeliveryResult{DeliveryType: "email_transcript", Target: strings.Join(recipients, ",")}, err
|
||||
}
|
||||
}
|
||||
return ActionDeliveryResult{DeliveryType: "email_transcript", Target: strings.Join(recipients, ","), ResponseBody: "queued", Queued: true}, nil
|
||||
}
|
||||
|
||||
subject, body, err := s.buildTranscriptEmail(ctx, accountID, conversationID)
|
||||
if err != nil {
|
||||
return ActionDeliveryResult{DeliveryType: "email_transcript", Target: strings.Join(recipients, ",")}, err
|
||||
@@ -546,6 +586,7 @@ func applyDeliveryResult(result *ActionExecutionResult, delivery ActionDeliveryR
|
||||
result.ResponseCode = delivery.ResponseCode
|
||||
result.ResponseBody = delivery.ResponseBody
|
||||
result.Retryable = delivery.Retryable
|
||||
result.Queued = delivery.Queued
|
||||
}
|
||||
|
||||
func firstStringParam(params map[string]interface{}, keys ...string) string {
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/gochat/gochat/internal/model"
|
||||
"github.com/gochat/gochat/internal/worker"
|
||||
)
|
||||
|
||||
type roundTripFunc func(*http.Request) (*http.Response, error)
|
||||
@@ -143,6 +144,99 @@ func TestActionService_SendEmailTranscript_DeliversSplitRecipients(t *testing.T)
|
||||
}
|
||||
}
|
||||
|
||||
func TestActionService_SendWebhookEvent_QueuesDurableDelivery(t *testing.T) {
|
||||
dbProvider := setupAutomationTestDBProvider(t)
|
||||
db := dbProvider.DB()
|
||||
accountID, _ := seedTestAccount(db, t)
|
||||
inboxID := seedTestInbox(db, t, accountID)
|
||||
contactID := seedTestContact(db, t, accountID)
|
||||
conversationID := seedTestConversation(db, t, accountID, inboxID, contactID)
|
||||
webhook := &recordingWebhookDeliverer{result: ActionDeliveryResult{DeliveryType: "webhook", Attempts: 1, ResponseCode: http.StatusOK}}
|
||||
restore := setAutomationActionDeliverersForTest(webhook, &recordingTranscriptDeliverer{})
|
||||
defer restore()
|
||||
|
||||
if err := db.Create(&model.Message{ConversationID: conversationID, AccountID: accountID, InboxID: inboxID, Content: "queued webhook", ContentType: "text", MessageType: "incoming"}).Error; err != nil {
|
||||
t.Fatalf("seed message: %v", err)
|
||||
}
|
||||
wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 12, 30, 0, 0, time.UTC) }))
|
||||
|
||||
result, err := NewActionServiceWithWorker(dbProvider, wp).ExecuteWithResult(context.Background(), accountID, conversationID, Action{
|
||||
ActionName: "send_webhook_event",
|
||||
ActionParams: map[string]interface{}{"url": "https://hooks.example/queued", "_event_name": "conversation_created"},
|
||||
}, ActionSourceAutomation, 99)
|
||||
if err != nil {
|
||||
t.Fatalf("queue webhook action: %v", err)
|
||||
}
|
||||
if len(webhook.requests) != 0 {
|
||||
t.Fatalf("webhook should not deliver synchronously, got %d requests", len(webhook.requests))
|
||||
}
|
||||
if !result.Queued || result.DeliveryType != "webhook" || result.Target != "https://hooks.example/queued" {
|
||||
t.Fatalf("unexpected queued webhook result: %#v", result)
|
||||
}
|
||||
|
||||
var count int64
|
||||
if err := db.Model(&model.BackgroundJob{}).Where("job_type = ? AND status = ?", TaskTypeAutomationWebhookDelivery, model.BackgroundJobStatusQueued).Count(&count).Error; err != nil {
|
||||
t.Fatalf("count webhook jobs: %v", err)
|
||||
}
|
||||
if count != 1 {
|
||||
t.Fatalf("expected one queued webhook job, got %d", count)
|
||||
}
|
||||
processed, err := wp.ProcessOne(context.Background())
|
||||
if err != nil || !processed {
|
||||
t.Fatalf("process webhook job: processed=%v err=%v", processed, err)
|
||||
}
|
||||
if len(webhook.requests) != 1 || webhook.requests[0].URL != "https://hooks.example/queued" || webhook.requests[0].EventName != "conversation_created" {
|
||||
t.Fatalf("unexpected durable webhook request: %#v", webhook.requests)
|
||||
}
|
||||
}
|
||||
|
||||
func TestActionService_SendEmailTranscript_QueuesDurableDeliveries(t *testing.T) {
|
||||
dbProvider := setupAutomationTestDBProvider(t)
|
||||
db := dbProvider.DB()
|
||||
accountID, _ := seedTestAccount(db, t)
|
||||
inboxID := seedTestInbox(db, t, accountID)
|
||||
contactID := seedTestContact(db, t, accountID)
|
||||
conversationID := seedTestConversation(db, t, accountID, inboxID, contactID)
|
||||
transcript := &recordingTranscriptDeliverer{result: ActionDeliveryResult{DeliveryType: "email_transcript", Attempts: 1}}
|
||||
restore := setAutomationActionDeliverersForTest(&recordingWebhookDeliverer{}, transcript)
|
||||
defer restore()
|
||||
|
||||
if err := db.Create(&model.Message{ConversationID: conversationID, AccountID: accountID, InboxID: inboxID, Content: "queued transcript", ContentType: "text", MessageType: "incoming"}).Error; err != nil {
|
||||
t.Fatalf("seed message: %v", err)
|
||||
}
|
||||
wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 12, 45, 0, 0, time.UTC) }))
|
||||
|
||||
result, err := NewActionServiceWithWorker(dbProvider, wp).ExecuteWithResult(context.Background(), accountID, conversationID, Action{
|
||||
ActionName: "send_email_transcript",
|
||||
ActionParams: map[string]interface{}{"email": "first@example.com, second@example.com"},
|
||||
}, ActionSourceAutomation, 99)
|
||||
if err != nil {
|
||||
t.Fatalf("queue transcript action: %v", err)
|
||||
}
|
||||
if len(transcript.requests) != 0 {
|
||||
t.Fatalf("transcript should not deliver synchronously, got %d requests", len(transcript.requests))
|
||||
}
|
||||
if !result.Queued || result.Target != "first@example.com,second@example.com" {
|
||||
t.Fatalf("unexpected queued transcript result: %#v", result)
|
||||
}
|
||||
|
||||
for i := 0; i < 2; i++ {
|
||||
processed, err := wp.ProcessOne(context.Background())
|
||||
if err != nil || !processed {
|
||||
t.Fatalf("process transcript job %d: processed=%v err=%v", i, processed, err)
|
||||
}
|
||||
}
|
||||
if len(transcript.requests) != 2 {
|
||||
t.Fatalf("expected two durable transcript deliveries, got %d", len(transcript.requests))
|
||||
}
|
||||
if transcript.requests[0].Recipient != "first@example.com" || transcript.requests[1].Recipient != "second@example.com" {
|
||||
t.Fatalf("unexpected durable transcript recipients: %#v", transcript.requests)
|
||||
}
|
||||
if !strings.Contains(transcript.requests[0].Body, "queued transcript") {
|
||||
t.Fatalf("expected durable transcript body to be rendered at job time: %#v", transcript.requests[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestAutomationRuleService_MatchAndExecute_RecordsEmailTranscriptFailureMetadata(t *testing.T) {
|
||||
dbProvider := setupAutomationTestDBProvider(t)
|
||||
db := dbProvider.DB()
|
||||
|
||||
@@ -45,6 +45,7 @@ func setupAutomationTestDB(t *testing.T) *gorm.DB {
|
||||
&Macro{},
|
||||
&MacroExecution{},
|
||||
&AutomationExecution{},
|
||||
&model.BackgroundJob{},
|
||||
&CsatSurveyResponse{},
|
||||
&ConversationLabel{},
|
||||
&ConversationMute{},
|
||||
|
||||
@@ -29,6 +29,7 @@ type ActionExecutionResult struct {
|
||||
ResponseCode int `json:"response_code,omitempty"`
|
||||
ResponseBody string `json:"response_body,omitempty"`
|
||||
Retryable bool `json:"retryable,omitempty"`
|
||||
Queued bool `json:"queued,omitempty"`
|
||||
}
|
||||
|
||||
// NewExecutionLogService creates a new ExecutionLogService.
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
|
||||
"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"
|
||||
)
|
||||
|
||||
@@ -34,6 +35,13 @@ func NewAutomationRuleListener(db DBProvider) *AutomationRuleListener {
|
||||
}
|
||||
}
|
||||
|
||||
func NewAutomationRuleListenerWithWorker(db DBProvider, wp *worker.WorkerPool) *AutomationRuleListener {
|
||||
return &AutomationRuleListener{
|
||||
db: db,
|
||||
ruleService: NewAutomationRuleServiceWithWorker(db, wp),
|
||||
}
|
||||
}
|
||||
|
||||
// Name returns the unique identifier for this listener.
|
||||
func (l *AutomationRuleListener) Name() string {
|
||||
return "automation_rule_listener"
|
||||
@@ -288,6 +296,12 @@ func RegisterAutomationRuleListener(dispatcher *channel.Dispatcher, db DBProvide
|
||||
applogger.L().Infof("registered automation rule listener with dispatcher")
|
||||
}
|
||||
|
||||
func RegisterAutomationRuleListenerWithWorker(dispatcher *channel.Dispatcher, db DBProvider, wp *worker.WorkerPool) {
|
||||
listener := NewAutomationRuleListenerWithWorker(db, wp)
|
||||
dispatcher.Register(listener)
|
||||
applogger.L().Infof("registered automation rule listener with durable worker")
|
||||
}
|
||||
|
||||
// String helper for event name validation
|
||||
func isValidAutomationEventName(name string) bool {
|
||||
validNames := map[string]bool{
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"strings"
|
||||
|
||||
"github.com/gochat/gochat/internal/model"
|
||||
"github.com/gochat/gochat/internal/worker"
|
||||
applogger "github.com/gochat/gochat/pkg/logger"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
@@ -14,7 +15,8 @@ import (
|
||||
// Reference: Chatwoot Macros::ExecutionService — same action handler pattern as AutomationRule,
|
||||
// but user-originated (supports 'self' assign). stamps user info.
|
||||
type MacroService struct {
|
||||
db DBProvider
|
||||
db DBProvider
|
||||
worker *worker.WorkerPool
|
||||
}
|
||||
|
||||
// NewMacroService creates a new MacroService.
|
||||
@@ -22,6 +24,17 @@ func NewMacroService(db DBProvider) *MacroService {
|
||||
return &MacroService{db: db}
|
||||
}
|
||||
|
||||
func NewMacroServiceWithWorker(db DBProvider, wp *worker.WorkerPool) *MacroService {
|
||||
s := NewMacroService(db)
|
||||
s.SetWorkerPool(wp)
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *MacroService) SetWorkerPool(wp *worker.WorkerPool) {
|
||||
s.worker = wp
|
||||
RegisterActionDeliveryJobs(wp, s.db)
|
||||
}
|
||||
|
||||
// GetByID retrieves a macro by ID.
|
||||
func (s *MacroService) GetByID(ctx context.Context, id uint) (*Macro, error) {
|
||||
var macro Macro
|
||||
@@ -162,7 +175,7 @@ func (s *MacroService) Execute(ctx context.Context, accountID uint, conversation
|
||||
|
||||
applogger.L().Infof("executing macro %d (%s) on conversation %d by user %d", macro.ID, macro.Name, conversationID, userID)
|
||||
|
||||
actionSvc := NewActionService(s.db)
|
||||
actionSvc := NewActionServiceWithWorker(s.db, s.worker)
|
||||
|
||||
for _, action := range macro.Actions {
|
||||
// Inject _source_user_id for "self" assignment support
|
||||
@@ -211,7 +224,7 @@ func (s *MacroService) ExecuteForDisplayIDs(ctx context.Context, accountID uint,
|
||||
return nil
|
||||
}
|
||||
|
||||
actionSvc := NewActionService(s.db)
|
||||
actionSvc := NewActionServiceWithWorker(s.db, s.worker)
|
||||
for _, conversation := range conversations {
|
||||
applogger.L().Infof("executing macro %d (%s) on conversation %d by user %d", macro.ID, macro.Name, conversation.ID, userID)
|
||||
for _, action := range macro.Actions {
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/gochat/gochat/internal/worker"
|
||||
applogger "github.com/gochat/gochat/pkg/logger"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
@@ -17,7 +18,8 @@ type DBProvider interface {
|
||||
// AutomationRuleService provides CRUD + condition matching + action execution for automation rules.
|
||||
// Reference: Chatwoot AutomationRules::ActionService + ConditionsFilterService pattern
|
||||
type AutomationRuleService struct {
|
||||
db DBProvider
|
||||
db DBProvider
|
||||
worker *worker.WorkerPool
|
||||
}
|
||||
|
||||
// NewAutomationRuleService creates a new AutomationRuleService.
|
||||
@@ -25,6 +27,17 @@ func NewAutomationRuleService(db DBProvider) *AutomationRuleService {
|
||||
return &AutomationRuleService{db: db}
|
||||
}
|
||||
|
||||
func NewAutomationRuleServiceWithWorker(db DBProvider, wp *worker.WorkerPool) *AutomationRuleService {
|
||||
s := NewAutomationRuleService(db)
|
||||
s.SetWorkerPool(wp)
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *AutomationRuleService) SetWorkerPool(wp *worker.WorkerPool) {
|
||||
s.worker = wp
|
||||
RegisterActionDeliveryJobs(wp, s.db)
|
||||
}
|
||||
|
||||
// GetByID retrieves an automation rule by ID.
|
||||
func (s *AutomationRuleService) GetByID(ctx context.Context, id uint) (*AutomationRule, error) {
|
||||
var rule AutomationRule
|
||||
@@ -233,7 +246,7 @@ func (s *AutomationRuleService) MatchAndExecute(ctx context.Context, accountID u
|
||||
return fmt.Errorf("failed to load conversation %d: %w", conversationID, err)
|
||||
}
|
||||
|
||||
actionSvc := NewActionService(s.db)
|
||||
actionSvc := NewActionServiceWithWorker(s.db, s.worker)
|
||||
logSvc := NewExecutionLogService(s.db)
|
||||
|
||||
for _, rule := range rules {
|
||||
|
||||
Reference in New Issue
Block a user