From 6851e20aa7e71358426a986104eec1620d128dec Mon Sep 17 00:00:00 2001 From: Rogee Date: Fri, 5 Jun 2026 17:13:26 +0800 Subject: [PATCH] feat(crm): queue contact exports --- docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md | 28 ++++++-- internal/app/bootstrap.go | 1 + internal/service/contact_export_worker.go | 43 ++++++++++++ internal/service/contact_service.go | 74 ++++++++++++++++++--- internal/service/contact_service_g3_test.go | 68 +++++++++++++++++++ internal/service/service_test_helper.go | 1 + 6 files changed, 201 insertions(+), 14 deletions(-) create mode 100644 internal/service/contact_export_worker.go diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index 0feaf0c7..752cc640 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(sla): queue applied sla scans`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(crm): queue contact exports`. - 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. 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. 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/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 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 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(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. | | `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. | @@ -1615,7 +1616,7 @@ Tracking table: | 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. | 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.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. | Review by `feat(crm): queue contact exports` | | 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 | @@ -1722,6 +1723,24 @@ env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache git diff --check ``` +P5.8 current checkpoint: + +- `contact:export` now mirrors Chatwoot `Account::ContactsExportJob` on the `low` queue. When a WorkerPool is configured, `POST /contacts/export` creates a pending `contact_exports` artifact and returns after enqueue instead of generating CSV synchronously on the request path. +- Worker replay reloads the export request from the persisted `column_names` and `filter_params`, generates the CSV artifact, stores `file_url`, `csv_data`, `row_count`, and `completed_at`, then creates the existing `contacts_export_complete` notification and invokes the fakeable mailer boundary. +- Completed export jobs are idempotent: a duplicate replay of an already-completed export is a no-op, so completion notifications and mail are not duplicated. +- Failed worker execution records the export error and leaves the background job retry/dead-letter state observable through `background_jobs`. +- No-worker construction still uses the synchronous fallback for focused tests and local paths that do not start the durable worker. +- Remaining Phase 5 work: delayed automation scheduled-item execution, WhatsApp/Twilio CSAT templates, provider webhooks/outbound delivery/status jobs, Captain/Copilot durable jobs, and conversation maintenance. + +P5.8 verification: + +```bash +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service -run 'ContactService_ExportContacts\|ContactExportJob' -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service ./internal/handler/api/v1 ./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. @@ -1918,3 +1937,4 @@ Verification milestone gates: - 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. +- 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. diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index b91a1fde..cadd6034 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -312,6 +312,7 @@ func Bootstrap(env string) (*App, error) { contactNoteService := service.NewContactNoteService(contactRepo, contactNoteRepo) contactService := service.NewContactService(contactRepo, contactInboxService, noteRepo) contactService.SetContactExportMailer(service.NewEnvContactExportMailer()) + contactService.SetWorkerPool(workerPool) // G4: Company service (depends on companyRepo, contactRepo, conversationRepo for nested queries) companyService := service.NewCompanyService(companyRepo, contactRepo, conversationRepo) diff --git a/internal/service/contact_export_worker.go b/internal/service/contact_export_worker.go new file mode 100644 index 00000000..ade07d18 --- /dev/null +++ b/internal/service/contact_export_worker.go @@ -0,0 +1,43 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "sync" + + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" +) + +const TaskTypeContactExport = "contact:export" + +type contactExportJob struct { + ExportID uint `json:"export_id"` +} + +var contactExportRegistrations sync.Map + +// RegisterContactExportJobs wires Chatwoot Account::ContactsExportJob into the +// durable worker. Handlers are registered once per WorkerPool instance. +func RegisterContactExportJobs(wp *worker.WorkerPool, svc *ContactService) { + if wp == nil || svc == nil { + return + } + if _, loaded := contactExportRegistrations.LoadOrStore(wp, struct{}{}); loaded { + return + } + wp.Register(TaskTypeContactExport, svc.performContactExportJob) +} + +func (s *ContactService) performContactExportJob(ctx context.Context, job *model.BackgroundJob) error { + var payload contactExportJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal contact export job: %w", err) + } + if payload.ExportID == 0 { + return fmt.Errorf("invalid contact export job payload: %#v", payload) + } + _, err := s.performContactExport(ctx, payload.ExportID) + return err +} diff --git a/internal/service/contact_service.go b/internal/service/contact_service.go index 612fade9..4375bf58 100644 --- a/internal/service/contact_service.go +++ b/internal/service/contact_service.go @@ -18,6 +18,7 @@ import ( "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/repository" "github.com/gochat/gochat/internal/search" + "github.com/gochat/gochat/internal/worker" applogger "github.com/gochat/gochat/pkg/logger" pkgvalidator "github.com/gochat/gochat/pkg/validator" ) @@ -31,6 +32,7 @@ type ContactService struct { searchIndexer SearchIndexer searchReader ContactSearchReader exportMailer ContactExportMailer + worker *worker.WorkerPool } // NewContactService creates a new Contact service. @@ -50,6 +52,11 @@ func (s *ContactService) SetContactExportMailer(mailer ContactExportMailer) { s.exportMailer = mailer } +func (s *ContactService) SetWorkerPool(wp *worker.WorkerPool) { + s.worker = wp + RegisterContactExportJobs(wp, s) +} + func (s *ContactService) indexContact(ctx context.Context, contact *model.Contact) { if s.searchIndexer != nil { logSearchIndexError("contact", contact.ID, s.searchIndexer.IndexContact(ctx, contact)) @@ -443,21 +450,53 @@ func (s *ContactService) ExportContacts(ctx context.Context, accountID, userID u if err := s.repo.DB().WithContext(ctx).Create(export).Error; err != nil { return nil, err } - if err := s.repo.DB().WithContext(ctx).Model(export).Update("status", string(model.DataImportStatusProcessing)).Error; err != nil { + if s.worker != nil { + _, err := s.worker.Enqueue(ctx, TaskTypeContactExport, contactExportJob{ExportID: export.ID}, worker.WithQueue("low"), worker.WithMaxAttempts(3), worker.WithIdempotencyKey(fmt.Sprintf("contact-export:%d", export.ID))) + if err != nil { + s.repo.DB().WithContext(ctx).Model(export).Updates(map[string]any{ + "status": string(model.DataImportStatusFailed), + "error": err.Error(), + }) + return export, err + } + return export, nil + } + return s.performContactExport(ctx, export.ID) +} + +func (s *ContactService) performContactExport(ctx context.Context, exportID uint) (*model.ContactExport, error) { + if !s.Ready() { + return nil, errors.New("contact service not ready") + } + + var export model.ContactExport + if err := s.repo.DB().WithContext(ctx).First(&export, exportID).Error; err != nil { + return nil, err + } + if export.Status == string(model.DataImportStatusCompleted) { + return &export, nil + } + + var account model.Account + if err := s.repo.DB().WithContext(ctx).First(&account, export.AccountID).Error; err != nil { + return nil, err + } + if err := s.repo.DB().WithContext(ctx).Model(&export).Update("status", string(model.DataImportStatusProcessing)).Error; err != nil { return nil, err } - csvData, rowCount, err := s.GenerateContactExportCSV(ctx, accountID, req) + req := contactExportRequestFromRecord(export) + csvData, rowCount, err := s.GenerateContactExportCSV(ctx, export.AccountID, req) if err != nil { - s.repo.DB().WithContext(ctx).Model(export).Updates(map[string]any{ + s.repo.DB().WithContext(ctx).Model(&export).Updates(map[string]any{ "status": string(model.DataImportStatusFailed), "error": err.Error(), }) - return export, err + return &export, err } completedAt := time.Now() - export.FileURL = fmt.Sprintf("/api/v1/accounts/%d/contacts/export/%d/download", accountID, export.ID) + export.FileURL = fmt.Sprintf("/api/v1/accounts/%d/contacts/export/%d/download", export.AccountID, export.ID) updates := map[string]any{ "status": string(model.DataImportStatusCompleted), "csv_data": csvData, @@ -465,19 +504,34 @@ func (s *ContactService) ExportContacts(ctx context.Context, accountID, userID u "file_url": export.FileURL, "completed_at": completedAt, } - if err := s.repo.DB().WithContext(ctx).Model(export).Updates(updates).Error; err != nil { + if err := s.repo.DB().WithContext(ctx).Model(&export).Updates(updates).Error; err != nil { return nil, err } - if err := s.repo.DB().WithContext(ctx).First(export, export.ID).Error; err != nil { + if err := s.repo.DB().WithContext(ctx).First(&export, export.ID).Error; err != nil { return nil, err } - if err := s.createContactExportNotification(ctx, export); err != nil { + if err := s.createContactExportNotification(ctx, &export); err != nil { applogger.L().Warnf("contact export notification failed: %v", err) } - if err := s.sendContactExportEmail(ctx, &account, export); err != nil { + if err := s.sendContactExportEmail(ctx, &account, &export); err != nil { applogger.L().Warnf("contact export email failed: %v", err) } - return export, nil + return &export, nil +} + +func contactExportRequestFromRecord(export model.ContactExport) ContactExportRequest { + var columnNames []string + if len(export.ColumnNames) > 0 { + _ = json.Unmarshal(export.ColumnNames, &columnNames) + } + var filters struct { + Payload []ContactExportFilterCondition `json:"payload"` + Label string `json:"label"` + } + if len(export.FilterParams) > 0 { + _ = json.Unmarshal(export.FilterParams, &filters) + } + return ContactExportRequest{ColumnNames: columnNames, Payload: filters.Payload, Label: filters.Label} } func (s *ContactService) sendContactExportEmail(ctx context.Context, account *model.Account, export *model.ContactExport) error { diff --git a/internal/service/contact_service_g3_test.go b/internal/service/contact_service_g3_test.go index 06390f72..c384d84a 100644 --- a/internal/service/contact_service_g3_test.go +++ b/internal/service/contact_service_g3_test.go @@ -14,6 +14,7 @@ import ( "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/search" + "github.com/gochat/gochat/internal/worker" ) type mockContactSearchReader struct { @@ -30,6 +31,7 @@ func (m *mockContactSearchReader) SearchContacts(ctx context.Context, accountID } type fakeContactExportMailer struct { + calls int called bool toEmail string fileURL string @@ -37,6 +39,7 @@ type fakeContactExportMailer struct { } func (m *fakeContactExportMailer) SendContactExportComplete(ctx context.Context, account *model.Account, user *model.User, export *model.ContactExport) error { + m.calls++ m.called = true m.toEmail = user.Email m.fileURL = export.FileURL @@ -268,6 +271,71 @@ func TestContactService_ExportContacts_PersistsArtifactAndNotification(t *testin assert.Equal(t, contactExportCompleteSubject, mailer.subject) } +func TestContactService_ExportContacts_QueuesDurableArtifactGeneration(t *testing.T) { + db, _, svc := setupContactService(t) + account := createTestAccount(t, db) + user := createTestUser(t, db, account.ID) + mailer := &fakeContactExportMailer{} + svc.SetContactExportMailer(mailer) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 18, 0, 0, 0, time.UTC) })) + svc.SetWorkerPool(wp) + contact := &model.Contact{AccountID: account.ID, Name: "Queued Alice", Email: "queued@example.com"} + require.NoError(t, db.Create(contact).Error) + + export, err := svc.ExportContacts(context.Background(), account.ID, user.ID, ContactExportRequest{ColumnNames: []string{"email"}}) + require.NoError(t, err) + assert.Equal(t, string(model.DataImportStatusPending), export.Status) + assert.Empty(t, export.CSVData) + assert.False(t, mailer.called) + + var jobCount int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ? AND queue = ? AND status = ?", TaskTypeContactExport, "low", model.BackgroundJobStatusQueued).Count(&jobCount).Error) + assert.Equal(t, int64(1), jobCount) + + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + + var completed model.ContactExport + require.NoError(t, db.First(&completed, export.ID).Error) + assert.Equal(t, string(model.DataImportStatusCompleted), completed.Status) + assert.Equal(t, 1, completed.RowCount) + assert.Contains(t, string(completed.CSVData), "email") + assert.Contains(t, string(completed.CSVData), "queued@example.com") + assert.Contains(t, completed.FileURL, fmt.Sprintf("/contacts/export/%d/download", completed.ID)) + assert.True(t, mailer.called) + assert.Equal(t, 1, mailer.calls) + assert.Equal(t, user.Email, mailer.toEmail) + + _, err = wp.Enqueue(context.Background(), TaskTypeContactExport, contactExportJob{ExportID: export.ID}, worker.WithQueue("low")) + require.NoError(t, err) + processed, err = wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + var notificationCount int64 + require.NoError(t, db.Model(&model.Notification{}).Where("primary_actor_type = ? AND primary_actor_id = ?", "ContactExport", export.ID).Count(¬ificationCount).Error) + assert.Equal(t, int64(1), notificationCount) + assert.Equal(t, 1, mailer.calls) +} + +func TestContactService_ContactExportJobRetriesMissingExport(t *testing.T) { + db, _, svc := setupContactService(t) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 18, 15, 0, 0, time.UTC) }), worker.WithBackoff(func(attempt int) time.Duration { return time.Minute })) + svc.SetWorkerPool(wp) + + _, err := wp.Enqueue(context.Background(), TaskTypeContactExport, contactExportJob{ExportID: 9999}, worker.WithQueue("low"), worker.WithMaxAttempts(3)) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + if err == nil || !processed { + t.Fatalf("expected missing contact export to retry, processed=%v err=%v", processed, err) + } + + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeContactExport).First(&job).Error) + assert.Equal(t, model.BackgroundJobStatusRetrying, job.Status) + assert.NotEmpty(t, job.LastError) +} + func TestContactService_ExportContacts_FiltersByLabelAndColumns(t *testing.T) { db, _, svc := setupContactService(t) account := createTestAccount(t, db) diff --git a/internal/service/service_test_helper.go b/internal/service/service_test_helper.go index 07dc4fe5..da18c83c 100644 --- a/internal/service/service_test_helper.go +++ b/internal/service/service_test_helper.go @@ -279,6 +279,7 @@ func setupContactServiceTestDB(t *testing.T) *gorm.DB { &model.ContactExport{}, &model.DataImport{}, &model.Notification{}, + &model.BackgroundJob{}, ); err != nil { t.Fatalf("failed to auto-migrate contact models: %v", err) }