diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index 752cc640..1c4dbff0 100644 --- a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md +++ b/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(crm): queue contact exports`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(conversations): queue maintenance 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.3 queues Meilisearch write-side index/delete jobs for conversations, messages, contacts, companies, and articles while keeping search reads Meilisearch-first; 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; P5.7 queues Chatwoot enterprise SLA account scans and applied-SLA evaluation jobs through the durable worker; P5.8 queues Chatwoot-style contact export artifact generation through durable `contact:export` jobs. Next active implementation slice is delayed automation actions, CSAT WhatsApp/Twilio templates, Captain/Copilot jobs, and remaining provider/conversation job fan-out. +- 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.3 queues Meilisearch write-side index/delete jobs for conversations, messages, contacts, companies, and articles while keeping search reads Meilisearch-first; 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; P5.7 queues Chatwoot enterprise SLA account scans and applied-SLA evaluation jobs through the durable worker; P5.8 queues Chatwoot-style contact export artifact generation through durable `contact:export` jobs; P5.12 queues scheduled item fan-out, one-off campaigns, snoozed conversation reopening, and account auto-resolution through durable jobs. Next active implementation slice is delayed automation actions, CSAT WhatsApp/Twilio templates, Captain/Copilot jobs, and remaining provider/outbound 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.3/P5.4/P5.5/P5.6/P5.7/P5.8 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT, SLA scan, and contact export cores are in Review; delayed automation/Captain/provider job integration remains open | +| Phase 5 | Background jobs and integrations | Doing | P5.1/P5.2/P5.3/P5.4/P5.5/P5.6/P5.7/P5.8/P5.12 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT, SLA scan, contact export, and conversation maintenance cores are in Review; delayed automation/Captain/provider 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(conversations): queue maintenance jobs` | Advances P5.12 and the scheduled-items portion of P5.5. `scheduled:trigger_items` now mirrors Chatwoot `TriggerScheduledItemsJob` by queueing due one-off campaigns, snoozed conversation reopening, account auto-resolution scheduling, and the next hourly root trigger. One-off campaigns claim active rows as `processing` before execution and complete after artifact creation, so duplicate worker replays do not duplicate campaign messages. | `go test ./internal/service -run 'ConversationMaintenanceJobs' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.12 section. | Move P5.12 conversation maintenance core to Review; continue update-message-status/bulk-action job depth plus CSAT channel templates, Captain/Copilot, provider webhook/outbound delivery, and analytics aggregation. | | `feat(crm): queue contact exports` | Advances P5.8 with a durable Chatwoot `Account::ContactsExportJob` equivalent. Contact export requests now create a pending `contact_exports` artifact and enqueue `contact:export` on the `low` queue when a WorkerPool is configured; worker replay generates the CSV artifact, marks completion, creates the persisted notification, and invokes the fakeable completion mailer. No-worker construction keeps the synchronous fallback for focused tests. | `go test ./internal/service -run 'ContactService_ExportContacts\|ContactExportJob' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.8 section. | Move P5.8 to Review; continue delayed automation scheduled items, CSAT channel templates, Captain/Copilot jobs, provider delivery, and conversation maintenance. | | `feat(sla): queue applied sla scans` | Advances P5.7 with the durable Chatwoot enterprise SLA chain. `sla:trigger_accounts` scans accounts with SLA policies, queues `sla:process_account` jobs, self-schedules the next five-minute scan, and `sla:process_account` queues `sla:process_applied` jobs for active/active_with_misses AppliedSLA rows; applied jobs call the existing idempotent `AppliedSlaService.Evaluate`. Bootstrap registers the handlers and seeds the first root scan. | `go test ./internal/service -run 'SlaProcessingJobs\|AppliedSlaService' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.7 section. | Move P5.7 to Review; continue delayed automation scheduled items, CSAT channel templates, Captain/Copilot jobs, provider delivery, and conversation maintenance. | | `feat(search): queue index updates durably` | Advances P5.3 with a durable `SearchIndexer` wrapper. Service-layer create/update/delete hooks now enqueue `search:index` jobs for conversations, messages, contacts, companies, and articles when a WorkerPool is configured; worker replay reloads the current account-scoped record before calling the real Meilisearch-backed `SearchService`, turns missing index records into delete jobs, and leaves search reads pointed at the live search service. | `go test ./internal/service -run 'DurableSearchIndexer\|SearchIndexHooks' -count=1`; `go test ./internal/service ./internal/search ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.3 section. | Move P5.3 to Review; continue delayed automation scheduled items and provider/template jobs. | @@ -1620,7 +1621,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. | Todo | | 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. | Todo | | P5.11 | Queue Captain document sync, crawl, response building, embeddings, and Copilot responses. | Captain document/crawl/response/embedding/Copilot jobs | `internal/service/captain_document_service.go`, Captain/Copilot services | Existing fakeable disabled/failure gates run under durable jobs; document statuses and Copilot message persistence survive worker restart. | Todo | -| P5.12 | Queue conversation maintenance jobs. | `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. | Todo | +| 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. | Doing: scheduled items, one-off campaigns, snooze reopen, and auto-resolution Review by `feat(conversations): queue maintenance jobs`; update-message-status and bulk-action jobs Todo | | P5.13 | Replace placeholder analytics/report builders that need background aggregation. | reporting jobs/services and report controllers | `internal/service/analytics_service.go`, reporting services | Frontend-visible reports no longer use placeholder values; any expensive aggregation is scheduled or cached with freshness rules. | Todo | P5.1 current checkpoint: @@ -1741,6 +1742,24 @@ env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache git diff --check ``` +P5.12 current checkpoint: + +- `scheduled:trigger_items` now mirrors Chatwoot `TriggerScheduledItemsJob` on the `scheduled_jobs` queue. Bootstrap registers the handlers and seeds the first root trigger; the root trigger self-schedules the next hourly run with an idempotency bucket. +- Root replay scans active due one-off campaigns in Chatwoot's three-day window and enqueues one durable `campaign:trigger_oneoff` job per campaign on the `low` queue. +- One-off campaign replay claims active campaigns as `processing` before execution, creates the campaign conversation/message through the existing campaign builder, marks the campaign completed, and no-ops duplicate replays once the campaign is no longer active. +- Root replay also queues `conversation:reopen_snoozed` and `account:conversations_resolution_scheduler`. The snooze job reopens due snoozed conversations and clears `snoozed_until`; the scheduler queues `conversation:resolution` jobs for accounts with auto-resolution enabled. +- Account resolution replay resolves old open conversations using Chatwoot-style minute thresholds from `auto_resolve_duration`, stamps `resolved_at`, and retries missing accounts through `background_jobs`. +- Remaining P5.12 work: `conversations/update_message_status_job.rb` depth and durable bulk-action fan-out. + +P5.12 verification: + +```bash +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service -run 'ConversationMaintenanceJobs' -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service ./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. @@ -1938,3 +1957,4 @@ Verification milestone gates: - 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. - 2026-06-05: P5.7 durable SLA scan checkpoint prepared as `feat(sla): queue applied sla scans`; the Chatwoot enterprise SLA chain now runs through `sla:trigger_accounts`, `sla:process_account`, and `sla:process_applied` jobs, self-schedules five-minute root scans, queues account/applied-SLA fan-out, and reuses idempotent `AppliedSlaService.Evaluate` for miss events and notifications. Focused SLA/worker/app tests passed. Next slice is delayed automation scheduled items, WhatsApp/Twilio CSAT templates, Captain/Copilot jobs, provider delivery, and conversation maintenance. - 2026-06-05: P5.8 durable contact export checkpoint prepared as `feat(crm): queue contact exports`; contact export requests now enqueue `contact:export` jobs on the low queue when a WorkerPool is configured, while worker replay generates the persisted CSV artifact, completion notification, and fakeable completion email. Focused service/API/worker/app tests passed. Next slice is delayed automation scheduled items, WhatsApp/Twilio CSAT templates, Captain/Copilot jobs, provider delivery, and conversation maintenance. +- 2026-06-05: P5.12 durable conversation maintenance checkpoint prepared as `feat(conversations): queue maintenance jobs`; the Chatwoot scheduled-items chain now runs through `scheduled:trigger_items`, due one-off campaigns queue `campaign:trigger_oneoff`, snoozed conversations reopen through `conversation:reopen_snoozed`, and account auto-resolution fans out through `account:conversations_resolution_scheduler` and `conversation:resolution`. Focused service/worker/app tests passed. Next slice is CSAT WhatsApp/Twilio templates, Captain/Copilot jobs, provider webhook/outbound delivery, update-message-status/bulk-action jobs, and analytics aggregation. diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index cadd6034..48f1aaa6 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -106,6 +106,10 @@ func Bootstrap(env string) (*App, error) { workerPool := worker.NewWorkerPool(db) automation.RegisterActionDeliveryJobs(workerPool, &dbProvider{db: db}) + service.RegisterConversationMaintenanceJobs(workerPool, db) + if _, err := service.EnqueueScheduledItemsTrigger(context.Background(), workerPool, time.Now()); err != nil { + applogger.L().Warnf("failed to enqueue initial scheduled item trigger: %v", err) + } // Step 5: Connect to Redis (ref: Chatwoot config/cable.yml) rdb, err := NewRedisClient(&cfg.Redis) diff --git a/internal/campaign/model.go b/internal/campaign/model.go index 31277c02..a15c5dbc 100644 --- a/internal/campaign/model.go +++ b/internal/campaign/model.go @@ -10,8 +10,9 @@ import ( type CampaignStatus string const ( - CampaignStatusActive CampaignStatus = "active" - CampaignStatusCompleted CampaignStatus = "completed" + CampaignStatusActive CampaignStatus = "active" + CampaignStatusCompleted CampaignStatus = "completed" + CampaignStatusProcessing CampaignStatus = "processing" ) // CampaignType represents the type of campaign. @@ -29,21 +30,21 @@ const ( // trigger_only_during_business_hours. type Campaign struct { model.Base - AccountID uint `gorm:"index;not null" json:"account_id"` - InboxID uint `gorm:"index;not null" json:"inbox_id"` - SenderID *uint `gorm:"index" json:"sender_id,omitempty"` - DisplayID uint `gorm:"uniqueIndex:idx_campaign_display;not null" json:"display_id"` - Title string `gorm:"size:255;not null" json:"title"` - Message string `gorm:"type:text;not null" json:"message"` - Description string `gorm:"type:text" json:"description"` - CampaignStatus CampaignStatus `gorm:"size:50;index;default:active" json:"campaign_status"` - CampaignType CampaignType `gorm:"size:50;not null" json:"campaign_type"` - Audience string `gorm:"type:jsonb;default:'{}'" json:"audience"` - TriggerRules string `gorm:"type:jsonb;default:'{}'" json:"trigger_rules"` - TemplateParams string `gorm:"type:jsonb;default:'{}'" json:"template_params"` - ScheduledAt *time.Time `gorm:"index" json:"scheduled_at,omitempty"` - Enabled bool `gorm:"default:true" json:"enabled"` - TriggerOnlyDuringBusinessHours bool `gorm:"default:false" json:"trigger_only_during_business_hours"` + AccountID uint `gorm:"index;not null" json:"account_id"` + InboxID uint `gorm:"index;not null" json:"inbox_id"` + SenderID *uint `gorm:"index" json:"sender_id,omitempty"` + DisplayID uint `gorm:"uniqueIndex:idx_campaign_display;not null" json:"display_id"` + Title string `gorm:"size:255;not null" json:"title"` + Message string `gorm:"type:text;not null" json:"message"` + Description string `gorm:"type:text" json:"description"` + CampaignStatus CampaignStatus `gorm:"size:50;index;default:active" json:"campaign_status"` + CampaignType CampaignType `gorm:"size:50;not null" json:"campaign_type"` + Audience string `gorm:"type:jsonb;default:'{}'" json:"audience"` + TriggerRules string `gorm:"type:jsonb;default:'{}'" json:"trigger_rules"` + TemplateParams string `gorm:"type:jsonb;default:'{}'" json:"template_params"` + ScheduledAt *time.Time `gorm:"index" json:"scheduled_at,omitempty"` + Enabled bool `gorm:"default:true" json:"enabled"` + TriggerOnlyDuringBusinessHours bool `gorm:"default:false" json:"trigger_only_during_business_hours"` } -func (Campaign) TableName() string { return "campaigns" } \ No newline at end of file +func (Campaign) TableName() string { return "campaigns" } diff --git a/internal/campaign/service.go b/internal/campaign/service.go index 805f061d..444e660c 100644 --- a/internal/campaign/service.go +++ b/internal/campaign/service.go @@ -115,6 +115,7 @@ func (b *CampaignConversationBuilder) Build(ctx context.Context, campaign *Campa AccountID: campaign.AccountID, InboxID: campaign.InboxID, ContactID: contactID, + CampaignID: &campaign.ID, Status: "open", ChannelType: "campaign", } @@ -154,4 +155,4 @@ func (b *CampaignConversationBuilder) Build(ctx context.Context, campaign *Campa } return nil -} \ No newline at end of file +} diff --git a/internal/service/conversation_maintenance_worker.go b/internal/service/conversation_maintenance_worker.go new file mode 100644 index 00000000..f65b635f --- /dev/null +++ b/internal/service/conversation_maintenance_worker.go @@ -0,0 +1,202 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "time" + + "github.com/gochat/gochat/internal/campaign" + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" + "gorm.io/gorm" +) + +const ( + TaskTypeScheduledTriggerItems = "scheduled:trigger_items" + TaskTypeCampaignTriggerOneoff = "campaign:trigger_oneoff" + TaskTypeConversationReopenSnoozed = "conversation:reopen_snoozed" + TaskTypeConversationResolutionScheduler = "account:conversations_resolution_scheduler" + TaskTypeConversationResolutionForAccount = "conversation:resolution" +) + +const ( + scheduledItemsInterval = time.Hour + scheduledItemsLookback = 3 * 24 * time.Hour + conversationResolutionLimit = 100 +) + +type campaignTriggerOneoffJob struct { + CampaignID uint `json:"campaign_id"` +} + +type conversationResolutionJob struct { + AccountID uint `json:"account_id"` +} + +var conversationMaintenanceRegistrations sync.Map + +// RegisterConversationMaintenanceJobs wires Chatwoot scheduled maintenance jobs +// into the durable worker: TriggerScheduledItemsJob fans out to campaign, +// snooze-reopen, and auto-resolution jobs. +func RegisterConversationMaintenanceJobs(wp *worker.WorkerPool, db *gorm.DB) { + registerConversationMaintenanceJobsWithNow(wp, db, time.Now) +} + +func registerConversationMaintenanceJobsWithNow(wp *worker.WorkerPool, db *gorm.DB, now func() time.Time) { + if wp == nil || db == nil { + return + } + if _, loaded := conversationMaintenanceRegistrations.LoadOrStore(wp, struct{}{}); loaded { + return + } + runner := &conversationMaintenanceRunner{wp: wp, db: db, now: now} + wp.Register(TaskTypeScheduledTriggerItems, runner.performScheduledTriggerItems) + wp.Register(TaskTypeCampaignTriggerOneoff, runner.performCampaignTriggerOneoff) + wp.Register(TaskTypeConversationReopenSnoozed, runner.performReopenSnoozed) + wp.Register(TaskTypeConversationResolutionScheduler, runner.performResolutionScheduler) + wp.Register(TaskTypeConversationResolutionForAccount, runner.performResolutionForAccount) +} + +func EnqueueScheduledItemsTrigger(ctx context.Context, wp *worker.WorkerPool, scheduledAt time.Time) (*model.BackgroundJob, error) { + if wp == nil { + return nil, nil + } + return wp.Enqueue(ctx, TaskTypeScheduledTriggerItems, nil, + worker.WithQueue("scheduled_jobs"), + worker.WithScheduledAt(scheduledAt), + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(scheduledItemsIdempotencyKey(scheduledAt)), + ) +} + +func scheduledItemsIdempotencyKey(scheduledAt time.Time) string { + bucket := scheduledAt.UTC().Truncate(scheduledItemsInterval).Unix() + return fmt.Sprintf("scheduled:trigger_items:%d", bucket) +} + +type conversationMaintenanceRunner struct { + wp *worker.WorkerPool + db *gorm.DB + now func() time.Time +} + +func (r *conversationMaintenanceRunner) performScheduledTriggerItems(ctx context.Context, job *model.BackgroundJob) error { + now := r.now() + var campaignIDs []uint + if err := r.db.WithContext(ctx).Model(&campaign.Campaign{}). + Where("campaign_type = ? AND campaign_status = ? AND enabled = ?", campaign.CampaignTypeOneOff, campaign.CampaignStatusActive, true). + Where("scheduled_at BETWEEN ? AND ?", now.Add(-scheduledItemsLookback), now). + Pluck("id", &campaignIDs).Error; err != nil { + return fmt.Errorf("find due one-off campaigns: %w", err) + } + for _, campaignID := range campaignIDs { + _, err := r.wp.Enqueue(ctx, TaskTypeCampaignTriggerOneoff, campaignTriggerOneoffJob{CampaignID: campaignID}, + worker.WithQueue("low"), + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(fmt.Sprintf("campaign:trigger_oneoff:%d", campaignID)), + ) + if err != nil { + return fmt.Errorf("enqueue campaign %d: %w", campaignID, err) + } + } + if _, err := r.wp.Enqueue(ctx, TaskTypeConversationReopenSnoozed, nil, worker.WithQueue("low"), worker.WithMaxAttempts(3)); err != nil { + return fmt.Errorf("enqueue reopen snoozed conversations: %w", err) + } + if _, err := r.wp.Enqueue(ctx, TaskTypeConversationResolutionScheduler, nil, worker.WithQueue("scheduled_jobs"), worker.WithMaxAttempts(3)); err != nil { + return fmt.Errorf("enqueue conversation resolution scheduler: %w", err) + } + _, err := EnqueueScheduledItemsTrigger(ctx, r.wp, now.Add(scheduledItemsInterval)) + return err +} + +func (r *conversationMaintenanceRunner) performCampaignTriggerOneoff(ctx context.Context, job *model.BackgroundJob) error { + var payload campaignTriggerOneoffJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal one-off campaign job: %w", err) + } + if payload.CampaignID == 0 { + return fmt.Errorf("invalid one-off campaign job payload: %#v", payload) + } + + claimed := false + if err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + result := tx.Model(&campaign.Campaign{}). + Where("id = ? AND campaign_type = ? AND campaign_status = ? AND enabled = ?", payload.CampaignID, campaign.CampaignTypeOneOff, campaign.CampaignStatusActive, true). + Update("campaign_status", campaign.CampaignStatusProcessing) + if result.Error != nil { + return result.Error + } + claimed = result.RowsAffected == 1 + return nil + }); err != nil { + return fmt.Errorf("claim one-off campaign %d: %w", payload.CampaignID, err) + } + if !claimed { + return nil + } + if err := campaign.NewCampaignService(r.db).TriggerCampaign(ctx, payload.CampaignID); err != nil { + _ = r.db.WithContext(ctx).Model(&campaign.Campaign{}).Where("id = ?", payload.CampaignID).Update("campaign_status", campaign.CampaignStatusActive).Error + return err + } + return r.db.WithContext(ctx).Model(&campaign.Campaign{}).Where("id = ?", payload.CampaignID).Update("campaign_status", campaign.CampaignStatusCompleted).Error +} + +func (r *conversationMaintenanceRunner) performReopenSnoozed(ctx context.Context, job *model.BackgroundJob) error { + now := r.now() + nowUnix := now.Unix() + lookbackUnix := now.Add(-scheduledItemsLookback).Unix() + updates := map[string]any{ + "status": string(model.ConversationStatusOpen), + "snoozed_until": nil, + "resumed_at": now, + } + return r.db.WithContext(ctx).Model(&model.Conversation{}). + Where("status = ?", string(model.ConversationStatusSnoozed)). + Where("snoozed_until BETWEEN ? AND ?", lookbackUnix, nowUnix). + Updates(updates).Error +} + +func (r *conversationMaintenanceRunner) performResolutionScheduler(ctx context.Context, job *model.BackgroundJob) error { + var accountIDs []uint + if err := r.db.WithContext(ctx).Model(&model.Account{}). + Where("auto_resolve_duration > 0"). + Pluck("id", &accountIDs).Error; err != nil { + return fmt.Errorf("find auto-resolve accounts: %w", err) + } + for _, accountID := range accountIDs { + if _, err := r.wp.Enqueue(ctx, TaskTypeConversationResolutionForAccount, conversationResolutionJob{AccountID: accountID}, worker.WithQueue("low"), worker.WithMaxAttempts(3)); err != nil { + return fmt.Errorf("enqueue account conversation resolution %d: %w", accountID, err) + } + } + return nil +} + +func (r *conversationMaintenanceRunner) performResolutionForAccount(ctx context.Context, job *model.BackgroundJob) error { + var payload conversationResolutionJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal conversation resolution job: %w", err) + } + if payload.AccountID == 0 { + return fmt.Errorf("invalid conversation resolution job payload: %#v", payload) + } + var account model.Account + if err := r.db.WithContext(ctx).First(&account, payload.AccountID).Error; err != nil { + return fmt.Errorf("load auto-resolve account %d: %w", payload.AccountID, err) + } + if account.AutoResolveDuration <= 0 { + return nil + } + cutoff := r.now().Add(-time.Duration(account.AutoResolveDuration) * time.Minute).Unix() + now := r.now() + updates := map[string]any{ + "status": string(model.ConversationStatusResolved), + "resolved_at": now, + } + return r.db.WithContext(ctx).Model(&model.Conversation{}). + Where("account_id = ? AND status = ? AND contact_id <> 0", account.ID, string(model.ConversationStatusOpen)). + Where("last_activity_at IS NOT NULL AND last_activity_at < ?", cutoff). + Limit(conversationResolutionLimit). + Updates(updates).Error +} diff --git a/internal/service/conversation_maintenance_worker_test.go b/internal/service/conversation_maintenance_worker_test.go new file mode 100644 index 00000000..b82a85fa --- /dev/null +++ b/internal/service/conversation_maintenance_worker_test.go @@ -0,0 +1,221 @@ +package service + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/gochat/gochat/internal/campaign" + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" + "gorm.io/gorm" +) + +func TestConversationMaintenanceJobsTriggerScheduledItemsFanOut(t *testing.T) { + now := time.Date(2026, 6, 5, 19, 0, 0, 0, time.UTC) + db := setupServiceTestDB(t) + if err := db.AutoMigrate(&campaign.Campaign{}); err != nil { + t.Fatalf("migrate campaign: %v", err) + } + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now })) + registerConversationMaintenanceJobsWithNow(wp, db, func() time.Time { return now }) + + account := createTestAccount(t, db) + inbox := createTestInbox(t, db, account.ID, "sms") + contact := createTestContact(t, db, account.ID) + dueAt := now.Add(-time.Hour) + futureAt := now.Add(time.Hour) + dueCampaign := createTestOneoffCampaign(t, db, account.ID, inbox.ID, contact.ID, dueAt) + createTestOneoffCampaign(t, db, account.ID, inbox.ID, contact.ID, futureAt) + + if _, err := EnqueueScheduledItemsTrigger(context.Background(), wp, now); err != nil { + t.Fatalf("enqueue scheduled trigger: %v", err) + } + processed, err := wp.ProcessOne(context.Background()) + if err != nil || !processed { + t.Fatalf("process scheduled trigger: processed=%v err=%v", processed, err) + } + + assertJobCount(t, db, TaskTypeCampaignTriggerOneoff, 1) + assertJobCount(t, db, TaskTypeConversationReopenSnoozed, 1) + assertJobCount(t, db, TaskTypeConversationResolutionScheduler, 1) + + var campaignJob model.BackgroundJob + if err := db.Where("job_type = ?", TaskTypeCampaignTriggerOneoff).First(&campaignJob).Error; err != nil { + t.Fatalf("load campaign job: %v", err) + } + if want := fmt.Sprintf("campaign:trigger_oneoff:%d", dueCampaign.ID); campaignJob.IdempotencyKey != want { + t.Fatalf("expected due campaign idempotency key %q, got %q", want, campaignJob.IdempotencyKey) + } + + var nextTrigger model.BackgroundJob + if err := db.Where("job_type = ? AND status = ?", TaskTypeScheduledTriggerItems, model.BackgroundJobStatusQueued).First(&nextTrigger).Error; err != nil { + t.Fatalf("load next trigger: %v", err) + } + if !nextTrigger.ScheduledAt.Equal(now.Add(scheduledItemsInterval)) { + t.Fatalf("expected next trigger at %s, got %s", now.Add(scheduledItemsInterval), nextTrigger.ScheduledAt) + } +} + +func TestConversationMaintenanceJobsProcessCampaignSnoozeAndResolution(t *testing.T) { + now := time.Date(2026, 6, 5, 20, 0, 0, 0, time.UTC) + db := setupServiceTestDB(t) + if err := db.AutoMigrate(&campaign.Campaign{}); err != nil { + t.Fatalf("migrate campaign: %v", err) + } + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now })) + registerConversationMaintenanceJobsWithNow(wp, db, func() time.Time { return now }) + + account := createTestAccount(t, db) + account.AutoResolveDuration = 30 + if err := db.Save(account).Error; err != nil { + t.Fatalf("save auto resolve account: %v", err) + } + inbox := createTestInbox(t, db, account.ID, "sms") + contact := createTestContact(t, db, account.ID) + dueCampaign := createTestOneoffCampaign(t, db, account.ID, inbox.ID, contact.ID, now.Add(-time.Hour)) + + dueSnooze := now.Add(-time.Minute).Unix() + snoozed := createTestConversation(t, db, account.ID, inbox.ID, contact.ID) + if err := db.Model(snoozed).Updates(map[string]any{"status": string(model.ConversationStatusSnoozed), "snoozed_until": dueSnooze}).Error; err != nil { + t.Fatalf("snooze conversation: %v", err) + } + oldActivity := now.Add(-45 * time.Minute).Unix() + oldOpen := createTestConversation(t, db, account.ID, inbox.ID, contact.ID) + if err := db.Model(oldOpen).Update("last_activity_at", oldActivity).Error; err != nil { + t.Fatalf("set old activity: %v", err) + } + recentActivity := now.Add(-5 * time.Minute).Unix() + recentOpen := createTestConversation(t, db, account.ID, inbox.ID, contact.ID) + if err := db.Model(recentOpen).Update("last_activity_at", recentActivity).Error; err != nil { + t.Fatalf("set recent activity: %v", err) + } + + if _, err := wp.Enqueue(context.Background(), TaskTypeCampaignTriggerOneoff, campaignTriggerOneoffJob{CampaignID: dueCampaign.ID}, worker.WithQueue("low")); err != nil { + t.Fatalf("enqueue campaign: %v", err) + } + if _, err := wp.Enqueue(context.Background(), TaskTypeConversationReopenSnoozed, nil, worker.WithQueue("low")); err != nil { + t.Fatalf("enqueue reopen: %v", err) + } + if _, err := wp.Enqueue(context.Background(), TaskTypeConversationResolutionScheduler, nil, worker.WithQueue("scheduled_jobs")); err != nil { + t.Fatalf("enqueue scheduler: %v", err) + } + + processRequiredJob(t, wp, "campaign") + processRequiredJob(t, wp, "reopen") + processRequiredJob(t, wp, "scheduler") + processRequiredJob(t, wp, "resolution") + + var completed campaign.Campaign + if err := db.First(&completed, dueCampaign.ID).Error; err != nil { + t.Fatalf("load campaign: %v", err) + } + if completed.CampaignStatus != campaign.CampaignStatusCompleted { + t.Fatalf("expected completed campaign, got %s", completed.CampaignStatus) + } + var campaignMessages int64 + if err := db.Model(&model.Message{}).Where("content = ?", dueCampaign.Message).Count(&campaignMessages).Error; err != nil { + t.Fatalf("count campaign messages: %v", err) + } + if campaignMessages != 1 { + t.Fatalf("expected one campaign message, got %d", campaignMessages) + } + + if _, err := wp.Enqueue(context.Background(), TaskTypeCampaignTriggerOneoff, campaignTriggerOneoffJob{CampaignID: dueCampaign.ID}, worker.WithQueue("low")); err != nil { + t.Fatalf("enqueue duplicate campaign: %v", err) + } + processRequiredJob(t, wp, "duplicate campaign") + if err := db.Model(&model.Message{}).Where("content = ?", dueCampaign.Message).Count(&campaignMessages).Error; err != nil { + t.Fatalf("count duplicate campaign messages: %v", err) + } + if campaignMessages != 1 { + t.Fatalf("expected duplicate campaign job to be idempotent, got %d messages", campaignMessages) + } + + var reopened model.Conversation + if err := db.First(&reopened, snoozed.ID).Error; err != nil { + t.Fatalf("load reopened conversation: %v", err) + } + if reopened.Status != string(model.ConversationStatusOpen) || reopened.SnoozedUntil != nil || reopened.ResumedAt == nil { + t.Fatalf("expected snoozed conversation reopened, got status=%s snoozed=%v resumed=%v", reopened.Status, reopened.SnoozedUntil, reopened.ResumedAt) + } + + var resolved model.Conversation + if err := db.First(&resolved, oldOpen.ID).Error; err != nil { + t.Fatalf("load resolved conversation: %v", err) + } + if resolved.Status != string(model.ConversationStatusResolved) || resolved.ResolvedAt == nil { + t.Fatalf("expected old open conversation resolved, got status=%s resolved_at=%v", resolved.Status, resolved.ResolvedAt) + } + var recent model.Conversation + if err := db.First(&recent, recentOpen.ID).Error; err != nil { + t.Fatalf("load recent conversation: %v", err) + } + if recent.Status != string(model.ConversationStatusOpen) { + t.Fatalf("expected recent conversation to remain open, got %s", recent.Status) + } +} + +func TestConversationMaintenanceJobsRetryMissingResolutionAccount(t *testing.T) { + now := time.Date(2026, 6, 5, 21, 0, 0, 0, time.UTC) + db := setupServiceTestDB(t) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now }), worker.WithBackoff(func(attempt int) time.Duration { return time.Minute })) + registerConversationMaintenanceJobsWithNow(wp, db, func() time.Time { return now }) + + if _, err := wp.Enqueue(context.Background(), TaskTypeConversationResolutionForAccount, conversationResolutionJob{AccountID: 9999}, worker.WithQueue("low"), worker.WithMaxAttempts(3)); err != nil { + t.Fatalf("enqueue missing account: %v", err) + } + processed, err := wp.ProcessOne(context.Background()) + if err == nil || !processed { + t.Fatalf("expected missing account to retry, processed=%v err=%v", processed, err) + } + var job model.BackgroundJob + if err := db.Where("job_type = ?", TaskTypeConversationResolutionForAccount).First(&job).Error; err != nil { + t.Fatalf("load resolution job: %v", err) + } + if job.Status != model.BackgroundJobStatusRetrying || job.LastError == "" { + t.Fatalf("expected retrying resolution job with error, got status=%s last_error=%q", job.Status, job.LastError) + } +} + +func createTestOneoffCampaign(t *testing.T, db *gorm.DB, accountID, inboxID, contactID uint, scheduledAt time.Time) *campaign.Campaign { + t.Helper() + c := &campaign.Campaign{ + AccountID: accountID, + InboxID: inboxID, + DisplayID: uint(time.Now().UnixNano()), + Title: fmt.Sprintf("Campaign %d", time.Now().UnixNano()), + Message: fmt.Sprintf("Campaign message %d", time.Now().UnixNano()), + CampaignStatus: campaign.CampaignStatusActive, + CampaignType: campaign.CampaignTypeOneOff, + Audience: fmt.Sprintf(`{"contact_ids":[%d]}`, contactID), + TriggerRules: `{}`, + TemplateParams: `{}`, + ScheduledAt: &scheduledAt, + Enabled: true, + } + if err := db.Create(c).Error; err != nil { + t.Fatalf("create campaign: %v", err) + } + return c +} + +func assertJobCount(t *testing.T, db *gorm.DB, jobType string, want int64) { + t.Helper() + var count int64 + if err := db.Model(&model.BackgroundJob{}).Where("job_type = ? AND status = ?", jobType, model.BackgroundJobStatusQueued).Count(&count).Error; err != nil { + t.Fatalf("count jobs %s: %v", jobType, err) + } + if count != want { + t.Fatalf("expected %d queued %s jobs, got %d", want, jobType, count) + } +} + +func processRequiredJob(t *testing.T, wp *worker.WorkerPool, name string) { + t.Helper() + processed, err := wp.ProcessOne(context.Background()) + if err != nil || !processed { + t.Fatalf("process %s job: processed=%v err=%v", name, processed, err) + } +}