diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index 929aa453..0feaf0c7 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(search): queue index updates durably`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(sla): queue applied sla scans`. - 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. Next active implementation slice is delayed automation actions, CSAT WhatsApp/Twilio templates, SLA scans, and remaining 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. Next active implementation slice is delayed automation actions, CSAT WhatsApp/Twilio templates, Captain/Copilot jobs, and remaining provider/conversation 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 durable worker, event dispatch, search indexing, automation delivery, macro, and CSAT cores are in Review; delayed automation/SLA/Captain 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 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT, and SLA scan 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,7 +101,8 @@ This ledger records the committed parity checkpoints that future slices should b | Commit | Scope | Verification summary | Follow-up state | | --- | --- | --- | --- | -| `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 SLA scans, delayed automation scheduled items, and provider/template jobs. | +| `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. | | `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. | @@ -1613,7 +1614,7 @@ Tracking table: | 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. | 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.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. | Review by `feat(sla): queue applied sla scans` | | 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 | | 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 | @@ -1703,6 +1704,24 @@ env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache git diff --check ``` +P5.7 current checkpoint: + +- `sla:trigger_accounts` mirrors Chatwoot enterprise `Sla::TriggerSlasForAccountsJob`: it scans accounts that have SLA policies and queues one `sla:process_account` job per account on the `medium` queue. +- The root trigger job runs on `scheduled_jobs`, uses five-minute idempotency buckets, and self-schedules the next scan after a successful run so bootstrap only needs to seed the first job. +- `sla:process_account` mirrors `Sla::ProcessAccountAppliedSlasJob`: it finds active and active_with_misses AppliedSLA rows for the account and queues `sla:process_applied` jobs. +- `sla:process_applied` mirrors `Sla::ProcessAppliedSlaJob`: it calls the existing `AppliedSlaService.Evaluate`, preserving idempotent miss events, notification fan-out, retry/backoff, and dead-letter visibility through `background_jobs`. +- Bootstrap registers the SLA job handlers and enqueues the initial root scan when the app boots. +- Remaining Phase 5 work: delayed automation scheduled-item execution, WhatsApp/Twilio CSAT templates, contact export artifacts, provider webhooks/outbound delivery/status jobs, Captain/Copilot durable jobs, and conversation maintenance. + +P5.7 verification: + +```bash +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service -run 'SlaProcessingJobs\|AppliedSlaService' -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. @@ -1898,3 +1917,4 @@ Verification milestone gates: - 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. +- 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. diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index 41c62bcb..b91a1fde 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -492,6 +492,10 @@ func Bootstrap(env string) (*App, error) { conversationService := service.NewConversationService(conversationRepo, messageRepo, channelDispatcher, inboxMemberService, accountUserRepo, teamRepo, teamMemberRepo) appliedSlaService := service.NewAppliedSlaService(appliedSlaRepo, slaEventRepo, slaPolicyRepo, conversationRepo) conversationService.SetAppliedSlaService(appliedSlaService) + service.RegisterSlaProcessingJobs(workerPool, db, appliedSlaService) + if _, err := service.EnqueueSlaAccountsScan(context.Background(), workerPool, time.Now()); err != nil { + applogger.L().Warnf("failed to enqueue initial SLA scan: %v", err) + } conversationParticipantService := service.NewConversationParticipantService(conversationParticipantRepo, conversationRepo) draftMessageService := service.NewDraftMessageService(draftMessageRepo, conversationRepo) inboxService := service.NewInboxService(inboxRepo, agentBotInboxRepo, agentBotRepo, campaignRepo, webhookSubRepo, waService, waRepo) diff --git a/internal/service/applied_sla_service_test.go b/internal/service/applied_sla_service_test.go index ad00c1b7..66a3dc3b 100644 --- a/internal/service/applied_sla_service_test.go +++ b/internal/service/applied_sla_service_test.go @@ -33,6 +33,7 @@ func setupAppliedSlaLifecycleTest(t *testing.T) (*AppliedSlaService, *gorm.DB) { &model.AppliedSLA{}, &model.SlaEvent{}, &model.Notification{}, + &model.BackgroundJob{}, )) t.Cleanup(func() { sqlDB, _ := db.DB() diff --git a/internal/service/sla_worker.go b/internal/service/sla_worker.go new file mode 100644 index 00000000..59d8ef9b --- /dev/null +++ b/internal/service/sla_worker.go @@ -0,0 +1,127 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "time" + + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" + "gorm.io/gorm" +) + +const ( + TaskTypeSlaTriggerAccounts = "sla:trigger_accounts" + TaskTypeSlaProcessAccount = "sla:process_account" + TaskTypeSlaProcessApplied = "sla:process_applied" +) + +const slaScanInterval = 5 * time.Minute + +type slaProcessAccountJob struct { + AccountID uint `json:"account_id"` +} + +type slaProcessAppliedJob struct { + AppliedSlaID uint `json:"applied_sla_id"` +} + +var slaProcessingRegistrations sync.Map + +// RegisterSlaProcessingJobs wires Chatwoot enterprise SLA scheduled jobs into +// the durable worker chain: +// TriggerSlasForAccountsJob -> ProcessAccountAppliedSlasJob -> ProcessAppliedSlaJob. +func RegisterSlaProcessingJobs(wp *worker.WorkerPool, db *gorm.DB, appliedSvc *AppliedSlaService) { + registerSlaProcessingJobsWithNow(wp, db, appliedSvc, time.Now) +} + +func registerSlaProcessingJobsWithNow(wp *worker.WorkerPool, db *gorm.DB, appliedSvc *AppliedSlaService, now func() time.Time) { + if wp == nil || db == nil || appliedSvc == nil { + return + } + if _, loaded := slaProcessingRegistrations.LoadOrStore(wp, struct{}{}); loaded { + return + } + runner := &slaProcessingJobRunner{wp: wp, db: db, appliedSvc: appliedSvc, now: now} + wp.Register(TaskTypeSlaTriggerAccounts, runner.performTriggerAccounts) + wp.Register(TaskTypeSlaProcessAccount, runner.performProcessAccount) + wp.Register(TaskTypeSlaProcessApplied, runner.performProcessApplied) +} + +// EnqueueSlaAccountsScan queues the root SLA scan job. Bootstrap uses this once +// at startup; the root job schedules the next scan after each successful run. +func EnqueueSlaAccountsScan(ctx context.Context, wp *worker.WorkerPool, scheduledAt time.Time) (*model.BackgroundJob, error) { + if wp == nil { + return nil, nil + } + return wp.Enqueue(ctx, TaskTypeSlaTriggerAccounts, nil, + worker.WithQueue("scheduled_jobs"), + worker.WithScheduledAt(scheduledAt), + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(slaTriggerIdempotencyKey(scheduledAt)), + ) +} + +func slaTriggerIdempotencyKey(scheduledAt time.Time) string { + bucket := scheduledAt.UTC().Truncate(slaScanInterval).Unix() + return fmt.Sprintf("sla:trigger_accounts:%d", bucket) +} + +type slaProcessingJobRunner struct { + wp *worker.WorkerPool + db *gorm.DB + appliedSvc *AppliedSlaService + now func() time.Time +} + +func (r *slaProcessingJobRunner) performTriggerAccounts(ctx context.Context, job *model.BackgroundJob) error { + var accountIDs []uint + if err := r.db.WithContext(ctx).Model(&model.SlaPolicy{}).Distinct("account_id").Pluck("account_id", &accountIDs).Error; err != nil { + return fmt.Errorf("find accounts with sla policies: %w", err) + } + for _, accountID := range accountIDs { + if _, err := r.wp.Enqueue(ctx, TaskTypeSlaProcessAccount, slaProcessAccountJob{AccountID: accountID}, worker.WithQueue("medium"), worker.WithMaxAttempts(3)); err != nil { + return fmt.Errorf("enqueue sla account %d: %w", accountID, err) + } + } + _, err := EnqueueSlaAccountsScan(ctx, r.wp, r.now().Add(slaScanInterval)) + return err +} + +func (r *slaProcessingJobRunner) performProcessAccount(ctx context.Context, job *model.BackgroundJob) error { + var payload slaProcessAccountJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal sla account job: %w", err) + } + if payload.AccountID == 0 { + return fmt.Errorf("invalid sla account job payload: %#v", payload) + } + + var appliedIDs []uint + statuses := []model.SLAStatus{model.SLAStatusActive, model.SLAStatusActiveWithMisses} + if err := r.db.WithContext(ctx).Model(&model.AppliedSLA{}). + Where("account_id = ? AND sla_status IN ?", payload.AccountID, statuses). + Pluck("id", &appliedIDs).Error; err != nil { + return fmt.Errorf("find active applied slas for account %d: %w", payload.AccountID, err) + } + for _, appliedID := range appliedIDs { + if _, err := r.wp.Enqueue(ctx, TaskTypeSlaProcessApplied, slaProcessAppliedJob{AppliedSlaID: appliedID}, worker.WithQueue("medium"), worker.WithMaxAttempts(3)); err != nil { + return fmt.Errorf("enqueue applied sla %d: %w", appliedID, err) + } + } + return nil +} + +func (r *slaProcessingJobRunner) performProcessApplied(ctx context.Context, job *model.BackgroundJob) error { + var payload slaProcessAppliedJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal applied sla job: %w", err) + } + if payload.AppliedSlaID == 0 { + return fmt.Errorf("invalid applied sla job payload: %#v", payload) + } + _, err := r.appliedSvc.Evaluate(ctx, payload.AppliedSlaID) + return err +} diff --git a/internal/service/sla_worker_test.go b/internal/service/sla_worker_test.go new file mode 100644 index 00000000..9b2a4c13 --- /dev/null +++ b/internal/service/sla_worker_test.go @@ -0,0 +1,120 @@ +package service + +import ( + "context" + "testing" + "time" + + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" +) + +func TestSlaProcessingJobsTriggerQueuesAccountScansAndReschedules(t *testing.T) { + svc, db := setupAppliedSlaLifecycleTest(t) + now := time.Date(2026, 6, 5, 16, 0, 0, 0, time.UTC) + _, _, _, _, _ = seedAppliedSlaConversation(t, db, now.Add(-time.Hour), model.SlaPolicy{FirstResponseTimeThreshold: 60}) + _ = createTestAccount(t, db) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now })) + registerSlaProcessingJobsWithNow(wp, db, svc, func() time.Time { return now }) + + if _, err := EnqueueSlaAccountsScan(context.Background(), wp, now); err != nil { + t.Fatalf("enqueue sla scan: %v", err) + } + processed, err := wp.ProcessOne(context.Background()) + if err != nil || !processed { + t.Fatalf("process trigger job: processed=%v err=%v", processed, err) + } + + var accountJobCount int64 + if err := db.Model(&model.BackgroundJob{}).Where("job_type = ? AND queue = ? AND status = ?", TaskTypeSlaProcessAccount, "medium", model.BackgroundJobStatusQueued).Count(&accountJobCount).Error; err != nil { + t.Fatalf("count account jobs: %v", err) + } + if accountJobCount != 1 { + t.Fatalf("expected one account scan job for accounts with SLA policies, got %d", accountJobCount) + } + + var nextTrigger model.BackgroundJob + if err := db.Where("job_type = ? AND status = ?", TaskTypeSlaTriggerAccounts, model.BackgroundJobStatusQueued).First(&nextTrigger).Error; err != nil { + t.Fatalf("load next trigger job: %v", err) + } + if !nextTrigger.ScheduledAt.Equal(now.Add(slaScanInterval)) { + t.Fatalf("expected next trigger at %s, got %s", now.Add(slaScanInterval), nextTrigger.ScheduledAt) + } +} + +func TestSlaProcessingJobsProcessAccountQueuesAndEvaluatesAppliedSlas(t *testing.T) { + svc, db := setupAppliedSlaLifecycleTest(t) + now := time.Date(2026, 6, 5, 16, 30, 0, 0, time.UTC) + account, _, _, policy, conversation := seedAppliedSlaConversation(t, db, time.Now().Add(-2*time.Hour), model.SlaPolicy{FirstResponseTimeThreshold: 60}) + applied, err := svc.CreateFromConversation(context.Background(), account.ID, conversation.ID, policy.ID) + if err != nil { + t.Fatalf("create applied sla: %v", err) + } + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now })) + registerSlaProcessingJobsWithNow(wp, db, svc, func() time.Time { return now }) + + if _, err := wp.Enqueue(context.Background(), TaskTypeSlaProcessAccount, slaProcessAccountJob{AccountID: account.ID}, worker.WithQueue("medium")); err != nil { + t.Fatalf("enqueue account job: %v", err) + } + processed, err := wp.ProcessOne(context.Background()) + if err != nil || !processed { + t.Fatalf("process account job: processed=%v err=%v", processed, err) + } + + var appliedJobCount int64 + if err := db.Model(&model.BackgroundJob{}).Where("job_type = ? AND status = ?", TaskTypeSlaProcessApplied, model.BackgroundJobStatusQueued).Count(&appliedJobCount).Error; err != nil { + t.Fatalf("count applied jobs: %v", err) + } + if appliedJobCount != 1 { + t.Fatalf("expected one applied SLA job, got %d", appliedJobCount) + } + + processed, err = wp.ProcessOne(context.Background()) + if err != nil || !processed { + t.Fatalf("process applied job: processed=%v err=%v", processed, err) + } + var reloaded model.AppliedSLA + if err := db.First(&reloaded, applied.ID).Error; err != nil { + t.Fatalf("reload applied sla: %v", err) + } + if reloaded.SLAStatus != model.SLAStatusActiveWithMisses { + t.Fatalf("expected SLA miss after durable evaluation, got %s", reloaded.SLAStatus) + } + + if _, err := wp.Enqueue(context.Background(), TaskTypeSlaProcessApplied, slaProcessAppliedJob{AppliedSlaID: applied.ID}, worker.WithQueue("medium")); err != nil { + t.Fatalf("enqueue duplicate applied job: %v", err) + } + processed, err = wp.ProcessOne(context.Background()) + if err != nil || !processed { + t.Fatalf("process duplicate applied job: processed=%v err=%v", processed, err) + } + var eventCount int64 + if err := db.Model(&model.SlaEvent{}).Where("applied_sla_id = ?", applied.ID).Count(&eventCount).Error; err != nil { + t.Fatalf("count sla events: %v", err) + } + if eventCount != 1 { + t.Fatalf("expected durable duplicate evaluation to stay idempotent, got %d events", eventCount) + } +} + +func TestSlaProcessingJobsRetryMissingAppliedSla(t *testing.T) { + svc, db := setupAppliedSlaLifecycleTest(t) + now := time.Date(2026, 6, 5, 17, 0, 0, 0, time.UTC) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now }), worker.WithBackoff(func(attempt int) time.Duration { return time.Minute })) + registerSlaProcessingJobsWithNow(wp, db, svc, func() time.Time { return now }) + + if _, err := wp.Enqueue(context.Background(), TaskTypeSlaProcessApplied, slaProcessAppliedJob{AppliedSlaID: 9999}, worker.WithQueue("medium"), worker.WithMaxAttempts(3)); err != nil { + t.Fatalf("enqueue missing applied job: %v", err) + } + processed, err := wp.ProcessOne(context.Background()) + if err == nil || !processed { + t.Fatalf("expected missing applied SLA to retry, processed=%v err=%v", processed, err) + } + var job model.BackgroundJob + if err := db.Where("job_type = ?", TaskTypeSlaProcessApplied).First(&job).Error; err != nil { + t.Fatalf("load applied job: %v", err) + } + if job.Status != model.BackgroundJobStatusRetrying || job.LastError == "" { + t.Fatalf("expected retrying applied SLA job with error, got status=%s last_error=%q", job.Status, job.LastError) + } +}