From 3a71001cb9c8fd0fb1081ee1060553caa9c3a094 Mon Sep 17 00:00:00 2001 From: Rogee Date: Sun, 7 Jun 2026 02:17:50 +0800 Subject: [PATCH] feat(crm): queue contact imports --- docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md | 43 ++++++++++---- internal/handler/api/v1/contact_handler.go | 2 +- .../handler/api/v1/contact_handler_g3_test.go | 2 +- internal/service/contact_import_worker.go | 47 ++++++++++++++++ internal/service/contact_service.go | 56 ++++++++++++++++--- internal/service/contact_service_g3_test.go | 56 +++++++++++++++++++ 6 files changed, 186 insertions(+), 20 deletions(-) create mode 100644 internal/service/contact_import_worker.go diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index b26c4889..3b59ecf0 100644 --- a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md +++ b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md @@ -49,11 +49,11 @@ Hermes task landing checklist: ## Current Baseline -- Current tracking checkpoint: 2026-06-07 Phase 6 placeholder audit refresh, prepared as `docs: refresh placeholder audit`. -- Latest implementation checkpoint: this checkpoint, prepared as `feat(reports): align first response distribution range`. -- Latest documentation/tooling checkpoint: this tracker update refreshes Phase 6 placeholder audit evidence and `docs/parity/frontend_smoke_report.md`; it confirms the broader placeholder scan has no reused-frontend critical account/contact/conversation/message/inbox/widget/public handler placeholder, with remaining `chatwootParityStub` use limited to public webhook nil-handler fallbacks. +- Current tracking checkpoint: 2026-06-07 P5.8b contact import DataImportJob parity, prepared as `feat(crm): queue contact imports`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(crm): queue contact imports`. +- Latest documentation/tooling checkpoint: this tracker update records durable contact import job parity from Chatwoot `contacts#import` plus `DataImportJob` behavior. No route artifacts change. - Plan landing status: complete for the current known Hermes plans and user-confirmed scope. Future work should update this file directly instead of opening a parallel tracker. -- Worktree status at this implementation checkpoint: v2 reports first-response distribution now matches inspected Chatwoot `V2::Reports::FirstResponseTimeDistributionBuilder` range semantics by allowing missing or partial `since/until` to omit the date filter while preserving range filtering when both params are present. P3.52 outgoing-message count builder parity, P3.51 inbox-label matrix filter parity, P3.50 bot metrics builder count parity, P3.49 reports conversation agent-metric parity, P3.48 live grouped team-scope parity, P3.47 grouped error-shape parity, P3.46 summary aggregate parity, P3.45 report bot metric timeseries count strategies, P3.44 live additive rollup parity, P3.43 rollup/backfill timezone parity, P3.42 timeseries timezone parity, P3.41 CSV/download parity, and P3.40 raw JSON/date parity for report, live-report, and summary-report endpoints remain implemented. Profile MFA, Facebook callbacks, account WhatsApp calls, onboarding, notification scoping, API v2 year-in-review, and WhatsApp calling toggle parity remain implemented from previous checkpoints. Live API/browser/enterprise smoke still needs the full PostgreSQL/Redis/Meilisearch/GoChat/Vite/Chrome stack. +- Worktree status at this implementation checkpoint: contact CSV imports now create pending `data_imports`, persist the uploaded CSV for worker replay, enqueue `contact:import` on the `low` queue when a WorkerPool is configured, and return immediately without creating contacts on the request path. Worker replay marks imports processing/completed/failed, updates record counters, and reuses existing CSV merge/label/custom-attribute behavior. Missing `import_file` now returns Chatwoot's exact `422 { error: "File is blank" }`. P3.53 first-response distribution parity and prior report/profile/channel/contact checkpoints remain implemented. Live API/browser/enterprise smoke still needs the full PostgreSQL/Redis/Meilisearch/GoChat/Vite/Chrome stack. - Next executable implementation checkpoint: continue Phase 2/3 drift audit for deeper report data-source/metric-builder drift or the next reused-frontend mismatch, or run B12 live smoke when the full PostgreSQL/Redis/Meilisearch/GoChat/Vite/Chrome stack is available. Re-run Phase 6 placeholder audit after future route/smoke changes. - `go test ./...` passes when run outside the restricted socket sandbox for the latest implementation baseline; the latest docs/tooling checkpoint verified `scripts/parity_frontend_smoke.sh --check` with workspace-local temp/cache dirs after `/tmp` was full. - Route dump succeeds with `967` registered routes after profile MFA route tracking. @@ -73,7 +73,7 @@ This section is the quick entry point for the next implementation turn. Use the | Tests and route artifacts | Required for every code checkpoint. Full `go test ./...` passes outside the restricted socket sandbox for the latest implementation baseline. | Run focused tests, full suite, `git diff --check`, and regenerate `docs/parity/*` only when route/tracked-set changes. | No checkpoint closes without verification or a documented environment blocker. | | Meilisearch search | In scope and mandatory. The Hermes search plan is landed into Phase 1/B6/P5.3; production DB search is not an acceptable final path. | Reopen only from live Meilisearch gate failure, stale index mutation path, or frontend search payload drift. | Meilisearch-first behavior is verified for global/entity search and indexing hooks. | | Enterprise paid features | In scope except SSO/SAML/LDAP/OIDC. SLA, Audit, CustomRole, AgentCapacity, Captain/Copilot, CSAT, InboxLimit, automation, macros, assignment policies, and limits remain tracked. | Use B7-B12 and Phase 5 rows as the source of truth; split new work only from fresh reference or smoke evidence. | Reused Chatwoot enterprise screens/API smoke without adapters. | -| Background jobs/integrations | Phase 5 is in Review with durable workers, dispatcher, search indexing, automation/macro/CSAT, SLA, exports, Captain/Copilot, maintenance, message status, bulk actions, and analytics covered by named checkpoints. | Keep closed unless a reference job path or live smoke exposes a missing durable side effect. | No frontend-visible async action silently succeeds without durable persistence or observable failure. | +| Background jobs/integrations | Phase 5 is in Review with durable workers, dispatcher, search indexing, automation/macro/CSAT, SLA, contact import/export, Captain/Copilot, maintenance, message status, bulk actions, and analytics covered by named checkpoints. | Keep closed unless a reference job path or live smoke exposes a missing durable side effect. | No frontend-visible async action silently succeeds without durable persistence or observable failure. | | Placeholder burn-down | Phase 6 is in Review; current audit shows only webhook nil-handler fallbacks still call `chatwootParityStub`. | Re-run `rg` placeholder audit after route/smoke changes and burn down any frontend-reachable stub. | No account/contact/conversation/message/inbox/widget/public critical stub is ownerless. | | Live B12 smoke | Harness exists; live run is environment-dependent. | Run when PostgreSQL, Redis, Meilisearch, GoChat, Vite, and Chrome are available. | `docs/parity/frontend_smoke_report.md` maps every failure to an owner slice or records clean pass. | @@ -146,7 +146,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 | Review | B7, B8, B9, B10, and B11 are in Review; B12 reused frontend smoke harnesses exist and optional live runs can expose follow-up slices | -| Phase 5 | Background jobs and integrations | Review | P5.1/P5.2/P5.3/P5.4/P5.5/P5.6/P5.7/P5.8/P5.9/P5.10/P5.11/P5.12 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT survey/template, SLA scan, contact export, inbound webhook persistence, outbound/provider delivery status, Captain document sync/crawl/response/embedding/Copilot/conversation responses, conversation maintenance, message status update, account bulk-action cores, P5.13 analytics rollups/timeseries, and B9.3 delayed-action reference closure are in Review; further work should be named drift from smoke/reference evidence | +| Phase 5 | Background jobs and integrations | Review | P5.1/P5.2/P5.3/P5.4/P5.5/P5.6/P5.7/P5.8/P5.8b/P5.9/P5.10/P5.11/P5.12 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT survey/template, SLA scan, contact export/import, inbound webhook persistence, outbound/provider delivery status, Captain document sync/crawl/response/embedding/Copilot/conversation responses, conversation maintenance, message status update, account bulk-action cores, P5.13 analytics rollups/timeseries, and B9.3 delayed-action reference closure are in Review; further work should be named drift from smoke/reference evidence | | Phase 6 | Core placeholder burn-down | Review | refreshed `docs/parity/placeholder_audit.md` shows only public webhook nil-handler fallbacks still call `chatwootParityStub`; no reused-frontend critical account/contact/conversation/message/inbox/widget/public path is ownerless | | 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 | @@ -156,6 +156,7 @@ This table is the shortest authoritative handoff view. If an older lower section | Priority | Workstream | Current state | Next checkpoint | Commit close rule | | --- | --- | --- | --- | --- | +| 0 | P5.8b contact import DataImportJob parity | Implemented for reused CRM import flow: GoChat now follows inspected Chatwoot `Api::V1::Accounts::ContactsController#import` missing-file error and moves successful imports behind a durable `contact:import` job when a WorkerPool is configured. The pending `data_imports` row carries the uploaded CSV bytes in `import_config` for worker replay until Go has a fuller ActiveStorage-equivalent attachment path. | Keep in Review; reopen from B12 CRM smoke or fresh reference evidence for ActiveStorage attachment metadata, CSV parser edge cases, import notification behavior, or exact `DataImport::ContactManager` validation drift. | Focused contact import service/handler tests, full `go test ./...`, and `git diff --check` must pass. No route artifacts change. | | 0 | P3.53 first-response distribution range parity | Implemented for reused v2 reports distribution clients: GoChat now follows inspected Chatwoot `V2::Reports::FirstResponseTimeDistributionBuilder` and `DateRangeHelper#range` by treating missing or partial `since/until` as no date filter, while still applying the half-open created-at range when both values are present. | Keep in Review; reopen from B12 reports smoke or fresh reference evidence for exact SQL aggregation behavior, invalid timestamp exceptions, or channel type bucketing drift. | Focused Analytics handler and service first-response distribution tests, full `go test ./...`, and `git diff --check` must pass. No route artifacts change. | | 0 | P3.52 outgoing messages count builder parity | Implemented for reused v2 reports outgoing-message clients: GoChat now follows inspected Chatwoot `Api::V2::Accounts::ReportsController#outgoing_messages_count` and `V2::Reports::OutgoingMessagesCountBuilder` by returning empty `422` for invalid or blank `group_by`, and by grouping label counts by tag name before resolving the current account's label id. This prevents local tag ids from leaking when taggings reference a same-named label outside the account. | Keep in Review; reopen from B12 reports smoke or fresh reference evidence for missing-date `range` behavior, output ordering, nil inbox id serialization, or ActsAsTaggableOn context drift. | Focused Analytics handler and service outgoing-message tests, full `go test ./...`, and `git diff --check` must pass. No route artifacts change. | | 0 | P3.51 inbox-label matrix filter parity | Implemented for reused v2 reports matrix clients: GoChat now follows inspected Chatwoot `V2::Reports::InboxLabelMatrixBuilder` by accepting optional `since/until`, `inbox_ids`, and `label_ids`; filtering the inbox and label axes before sorting by name/title; and counting only matching conversation labels whose conversations fall in the optional range and selected inboxes. | Keep in Review; reopen from B12 reports smoke or fresh reference evidence for ActsAsTaggableOn title/name divergence, array param encoding drift, or optional invalid-date controller behavior. | Focused Analytics handler and service matrix tests, full `go test ./...`, and `git diff --check` must pass. No route artifacts change. | @@ -227,6 +228,7 @@ These rows are the executable development plan from this point forward. A checkp | P3.23 nested contact inbox creation parity | `internal/handler/api/v1/contact_handler.go`, `internal/service/contact_inbox_service.go`, `internal/repository/contact_inbox_repo.go`, `internal/handler/api/v1/crm_serializer.go`, `internal/router/router.go`, `cmd/route_parity/main.go`, nested contact inbox handler/service/repository tests | `reference/chatwoot/config/routes.rb:212`, `reference/chatwoot/app/controllers/api/v1/accounts/contacts/contact_inboxes_controller.rb`, `reference/chatwoot/app/controllers/concerns/hmac_concern.rb`, `reference/chatwoot/app/builders/contact_inbox_builder.rb`, `reference/chatwoot/app/views/api/v1/accounts/contacts/contact_inboxes/create.json.jbuilder`, `reference/chatwoot/app/views/api/v1/models/_contact_inbox.json.jbuilder`, `reference/chatwoot/spec/controllers/api/v1/accounts/contacts/contact_inboxes_controller_spec.rb` | Done. Chatwoot `ContactInboxBuilder` behavior is implemented for `POST /api/v1/accounts/:account_id/contacts/:contact_id/contact_inboxes`: raw JSON/form/query params provide `inbox_id`, optional `source_id`, and `hmac_verified`; contact and inbox resolution is account-scoped; missing source IDs are generated by supported channel (`api`/`web_widget` UUID, email, sms phone, whatsapp phone without `+`, twilio sms/whatsapp medium); existing contact+inbox+source rows are returned idempotently; `hmac_verified` is set on creation; tokens are generated; inbox is preloaded; and the response is only `{ source_id, inbox: inbox_slim }`. | Review by `feat(contacts): align contact inbox creation`; focused handler tests cover raw payload shape, HMAC creation, generated source IDs, email idempotency, cross-account inbox rejection, and missing-phone Twilio failure; service tests cover WhatsApp/Twilio generation and idempotency; repository tests cover contact+inbox+source lookup; route parity, full `go test ./...`, and `git diff --check` passed. | | P3.2a invitation/confirmation mail parity | `internal/service/profile_service.go`, `internal/service/profile_confirmation_mailer.go`, `internal/handler/api/v1/profile_handler.go`, `internal/service/agent_service.go`, `internal/repository/agent_repo.go`, `internal/model/user.go`, `migrations/000032_add_users_unconfirmed_email.*`, profile/agent handler tests | `reference/chatwoot/app/controllers/api/v1/accounts/agents_controller.rb`, `reference/chatwoot/app/builders/agent_builder.rb`, `reference/chatwoot/app/views/devise/mailer/confirmation_instructions.html.erb`, `reference/chatwoot/spec/mailers/confirmation_instructions_spec.rb`, `reference/chatwoot/spec/enterprise/mailers/devise_mailer_spec.rb` for non-SAML invitation wording only | Done. A shared fakeable confirmation mailer builds Chatwoot-shaped confirmation/invitation payloads; profile resend persists confirmation/reset tokens and delivers no-op/confirmation/invitation states; newly created invited agents get workspace invitation mail; `unconfirmed_email` is modeled for email-update branch routing; environment SMTP remains a no-op when not configured. SSO/SAML/LDAP/OIDC mail variants stay excluded. | Review by `feat(profile): send confirmation invitations`; focused tests cover confirmed no-op, normal confirmation mail, invited workspace invitation mail, agent creation/inviter context, hashed reset-token persistence, and no network in default tests; full `go test ./...` and `git diff --check` passed. | | P5.11a Captain document crawl/schedule | `internal/service/captain_document_service.go`, `internal/service/captain_document_worker.go`, `internal/app/bootstrap.go` | `reference/chatwoot/enterprise/app/jobs/captain/documents/crawl_job.rb`, `schedule_syncs_job.rb`, `perform_sync_job.rb`, Firecrawl/simple parser jobs | Durable schedule/crawl producers and handlers with fakeable crawl/parser boundaries. Missing provider config is a failed `crawl_disabled` state, not placeholder success. | Review by `feat(captain): queue document crawl jobs`; focused worker tests prove enqueue, replay, account scope, idempotent scheduler, and disabled/failure states. | +| P5.8b contact import DataImportJob parity | `internal/service/contact_service.go`, `internal/service/contact_import_worker.go`, `internal/handler/api/v1/contact_handler.go`, contact import service/handler tests | `reference/chatwoot/app/controllers/api/v1/accounts/contacts_controller.rb#import`, `reference/chatwoot/spec/controllers/api/v1/accounts/contacts_controller_spec.rb`, `reference/chatwoot/app/jobs/data_import_job.rb`, `reference/chatwoot/app/services/data_import/contact_manager.rb` | Contact import now follows Chatwoot's durable import boundary: missing `import_file` returns `422 { error: "File is blank" }`; successful imports create pending `data_imports`, persist the uploaded CSV for replay in `import_config`, enqueue `contact:import` on the `low` queue when a WorkerPool exists, and worker replay transitions processing/completed/failed while reusing the CSV merge, label validation, and custom-attribute logic. | Review by `feat(crm): queue contact imports`; focused contact import service/handler tests, full `go test ./...`, and `git diff --check` must pass. No route artifacts change. | | P5.11b Captain response/embedding fan-out | Captain document/assistant-response services and repositories, Meilisearch/embedding boundaries | `response_builder_job.rb`, `enterprise/app/jobs/captain/llm/update_embedding_job.rb`, FAQ generator/embedding services | Queue FAQ response generation after successful document content changes, reset unedited responses, create/update assistant responses, and fan out embedding update work behind fakeable LLM gates. | Review by `feat(captain): queue response embedding jobs`; tests cover response reset/create, embedding-disabled retry, fake embedding success, account scope, and no external network in default tests. | | P5.11c Copilot and conversation response jobs | `internal/service/copilot_service.go`, `internal/service/copilot_response_worker.go`, `internal/service/captain_conversation_service.go`, `internal/service/message_service.go`, `internal/app/bootstrap.go` | `enterprise/app/jobs/captain/copilot/response_job.rb`, `enterprise/app/jobs/captain/conversation/response_builder_job.rb`, `enterprise/app/services/captain/copilot/chat_service.rb`, `enterprise/app/services/enterprise/message_templates/hook_execution_service.rb`, `enterprise/app/models/copilot_message.rb` | Queue assistant replies after Copilot user messages and Captain pending-conversation triggers. Persist assistant messages, enqueue Captain conversation replies/handoff messages, open handoff conversations, and keep fakeable provider disabled/failure states observable through durable retry. | Review by `feat(captain): queue copilot response jobs`; focused tests cover Copilot enqueue/persist/fallback/retry, Captain conversation enqueue/handoff/retry/non-pending skip, and service/worker/app package replay. | | P5.13a analytics placeholder burn-down | `internal/service/analytics_service.go`, `internal/service/analytics_query_helpers.go`, live/report handlers | Chatwoot `live_reports_controller.rb`, `reports_controller.rb`, `BotMetricsBuilder`, `InboxLabelMatrixBuilder`, `FirstResponseTimeDistributionBuilder`, `OutgoingMessagesCountBuilder` | Replace frontend-visible zero/empty placeholder responses for live conversations, grouped live conversations, bot summary/metrics, conversation summary, inbox-label matrix, first-response distribution, and outgoing-message counts with persisted conversation/message/reporting-event queries. | Review by `feat(reports): derive analytics aggregates`; focused service/handler tests prove non-zero values from persisted rows and `rg` finds no placeholder TODOs in these methods. | @@ -405,6 +407,7 @@ This ledger records the committed parity checkpoints that future slices should b | `feat(automation): close delayed action parity` | Closes B9.3 for the current reference. `reference/chatwoot` automation models, action service, enterprise action service, listener, and dashboard action components expose no configurable delayed action params, so no separate delayed automation scheduler is required beyond P5.12 scheduled items. The checkpoint completes remaining automation action side effects: `send_email_to_team` now queues durable `automation:team_email_delivery` jobs and replays through the fakeable mailer boundary, while enterprise `add_sla` attaches an account-scoped SLA policy and creates idempotent AppliedSLA targets. The automation serializer now preserves Chatwoot's object-shaped `send_email_to_team` params. | `go test ./internal/automation -run 'ActionService\|AutomationRuleService_MatchAndExecute\|ExecutionLogService' -count=1`; `go test ./internal/handler/api/v1 -run AutomationRule -count=1`; `go test ./internal/automation ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`. | B9.3 and Phase 5 move to Review; continue Phase 2/3 drift audit, Phase 6 placeholder audit, and optional B12 live smoke. | | `feat(reports): add analytics timeseries rollups` | Completes P5.13b for scheduled/cached analytics parity. `GET /api/v1/accounts/:account_id/reports` and v2 `/reports` now route to metric timeseries instead of summary, with support for account/inbox/agent/team/label dimensions, day/hour/week/month/year buckets, conversation/message/reporting-event metrics, and business-hours averages. Analytics summary/dimension/traffic reads call `EnsureRollupsForRange` for lazy freshness; `reporting:rollup_day` jobs provide durable day recompute; rollup replacement uses `Unscoped` delete so soft-deleted rows cannot violate uniqueness on recompute. | `go test ./internal/service -run 'Analytics' -count=1`; `go test ./internal/handler/api/v1 -run 'Analytics\|LiveReport' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/router ./internal/worker ./internal/app -count=1`; `go run ./cmd/dump_routes > docs/parity/gochat_routes.txt`; `go run ./cmd/route_parity`; `go test ./...`; `git diff --check`; full verification recorded in the P5.13 section. | P5.13 moves to Review; B9.3 is completed by `feat(automation): close delayed action parity`; continue Phase 2/3 drift and Phase 6 placeholder audits. | | `feat(reports): derive analytics aggregates` | Advances P5.13a by replacing frontend-visible analytics placeholder responses with persisted aggregations. Live report conversation metrics now count open/unattended/unassigned/pending conversations with team filtering; grouped live reports return assignee/team grouped counts; bot summary/metrics, conversation summary, inbox-label matrix, first-response distribution, and outgoing-message counts are derived from conversations, messages, labels, agent-bot bindings, and reporting events instead of fixed zero/empty JSON. | `go test ./internal/service -run 'Analytics' -count=1`; `go test ./internal/handler/api/v1 -run 'Analytics\|LiveReport' -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.13 section. | P5.13a moves to Review; continue P5.13b scheduled/cached rollup freshness and `/reports` timeseries index parity, then B9.3 delayed automation action check. | +| `feat(crm): queue contact imports` | Advances P5.8b with a durable Chatwoot `DataImportJob` contacts import equivalent. `POST /contacts/import` now returns Chatwoot's exact missing-file error, successful imports create a pending `data_imports` row with replayable CSV payload, enqueue `contact:import` on the `low` queue when a WorkerPool is configured, and process rows asynchronously through the existing CSV merge/label/custom-attribute importer. No-worker construction keeps the synchronous fallback for focused tests. | `go test ./internal/service -run 'ContactService_ImportContacts\|ContactService_ContactImport\|ContactService_ExportContacts_Queues\|ContactService_ContactExport' -count=1`; `go test ./internal/handler/api/v1 -run 'ContactImport\|ContactHandlerCRUDTestSuite/TestImport' -count=1`; `go test ./internal/service -run 'Contact' -count=1`; `go test ./internal/handler/api/v1 -run 'Contact' -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.8b section. | Move P5.8b to Review; continue Phase 2/3 drift audit, Phase 6 placeholder audit, or B12 live smoke. | | `feat(captain): queue copilot response jobs` | Completes P5.11c with durable Chatwoot `Captain::Copilot::ResponseJob` and `Captain::Conversation::ResponseBuilderJob` equivalents. Copilot thread/message creation now persists the user message and enqueues `captain:copilot_response` when a WorkerPool is configured, while worker replay reloads the account/user/thread/message scope and persists assistant replies through a fakeable backend or the existing no-provider fallback. Incoming pending conversation messages for Captain-enabled inboxes now enqueue `captain:conversation_response_builder`; replay collects public incoming/outgoing history, creates Captain outgoing replies, enqueues provider send-reply, and opens the conversation with a handoff message when the backend requests handoff. | `go test ./internal/service -run 'CopilotResponse\|CaptainConversation\|MessageService' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.11 section. | P5.11 moves to Review; continue P5.13 analytics aggregation, then Phase 2/3 drift and Phase 6 placeholder audits as smoke exposes gaps. | | `feat(captain): queue response embedding jobs` | Advances P5.11b with durable Captain document response building and embedding update fan-out. Successful document sync/parser content updates enqueue `captain:document_response_builder`, worker replay resets only unedited document responses, preserves edited responses, creates approved `Captain::Document` assistant responses from a fakeable FAQ backend, and enqueues `captain:llm_update_embedding` jobs for created responses. Embedding replay reloads account-scoped responses, uses a fakeable embedding backend or configured LLM provider, and surfaces missing provider config as retryable worker failures. | `go test ./internal/service -run 'CaptainDocumentService\|EnqueueCaptainDocumentScheduleSyncs' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.11 section. | P5.11b moves to Review; continue P5.11c Copilot/conversation response jobs, then P5.13 analytics aggregation. | | `feat(captain): queue document crawl jobs` | Advances P5.11a with durable Captain document crawl, simple-page parser, and auto-sync scheduler jobs. Document create/crawl requests enqueue `captain:document_crawl`, crawl replay uses a fakeable crawl backend to enqueue normalized `captain:document_page_crawl_parse` jobs, parser replay creates or updates account-scoped document content through a fakeable parser backend, and the `captain:documents_schedule_syncs` root job scans stale synced/failed/syncing documents for accounts with `captain_document_auto_sync` enabled before enqueueing `captain:document_sync` work with daily idempotency. Bootstrap seeds the scheduler after Captain document services are wired. | `go test ./internal/service -run 'CaptainDocumentService\|EnqueueCaptainDocumentScheduleSyncs' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.11 section. | P5.11a moves to Review; continue P5.11b response/embedding fan-out, then P5.11c Copilot/conversation response jobs. | @@ -603,7 +606,7 @@ Current B4 contacts/companies checkpoint: - Added `DELETE /contacts/:contact_id/avatar`, `DELETE /companies/:company_id/avatar`, and `POST /companies/:company_id/destroy_custom_attributes` with Chatwoot-shaped `{ payload }` responses and `422 { error: "custom_attributes must be an array" }` validation. - Added trailing-slash aliases for Chatwoot nested label/contact-inbox collection routes. - Added exact `POST /api/v1/accounts/:account_id/actions/contact_merge` and raw contact-model response shape for dashboard merge actions; merge moves conversations, contact inboxes, notes, and contact-sender messages to the base contact while preserving base attributes over mergee attributes. -- Added Chatwoot-shaped `data_imports` persistence for contact CSV imports. Missing `import_file` now returns `422 { error: "failed to import contacts" }`, successful imports return empty `200 OK`, import state/count/error fields are recorded, and CSV rows merge/update existing contacts by identifier/email/phone while preserving labels and unknown/custom columns in contact attributes. +- Added Chatwoot-shaped `data_imports` persistence for contact CSV imports. Missing `import_file` now returns `422 { error: "File is blank" }`; successful imports create pending replayable data imports and enqueue `contact:import` when workers are configured; import state/count/error fields are recorded; and CSV rows merge/update existing contacts by identifier/email/phone while preserving labels and unknown/custom columns in contact attributes. - Added persisted contact export artifacts for `POST /contacts/export`: default/requested column handling, label/filter export inputs, UTF-8 BOM CSV content, local download URL, row counts, and a user notification carrying file metadata. - Added Chatwoot note serializer parity for contact notes: raw array/object responses, nested agent payloads, flat and nested note request bodies, empty `200 OK` destroy, hard-delete semantics, and `PUT`/`PATCH` update routes. Company note payloads now include nested user metadata. - Added nested contact/company conversation serializer parity: both endpoints now return `{ payload: [...] }` arrays built from the shared Chatwoot conversation partial, including nested sender metadata, latest message payloads, `last_non_activity_message`, labels, timestamps, priority, and unread fields. Company conversations are limited to the latest 20 by `last_activity_at DESC` like the enterprise controller. @@ -1936,14 +1939,14 @@ Reference job families to keep in scope: - Core dispatcher jobs: `reference/chatwoot/app/jobs/event_dispatcher_job.rb`, `action_cable_broadcast_job.rb`, `hook_job.rb`, `webhook_job.rb`, and provider webhook/event jobs. - Conversation/message jobs: `send_reply_job.rb`, `bulk_actions_job.rb`, `macros_execution_job.rb`, `conversations/resolution_job.rb`, `conversations/reopen_snoozed_conversations_job.rb`, `conversations/update_message_status_job.rb`, and activity/mention jobs. -- CRM/import/export jobs: `account/contacts_export_job.rb`, company/avatar/label jobs, and migration/backfill jobs that affect frontend-visible data. +- CRM/import/export jobs: `data_import_job.rb`, `account/contacts_export_job.rb`, company/avatar/label jobs, and migration/backfill jobs that affect frontend-visible data. - Enterprise SLA jobs: `enterprise/app/jobs/sla/trigger_slas_for_accounts_job.rb`, `process_account_applied_slas_job.rb`, and `process_applied_sla_job.rb`. - Captain/Copilot jobs: document schedule/perform/crawl/response-builder jobs, embedding update jobs, conversation response builder, and Copilot response jobs. - Channel template/status jobs: WhatsApp/Twilio template sync, CSAT template delivery, delivery status, and provider-specific outbound/inbound jobs. Known hotspots: -- `internal/worker/worker.go` has the committed P5.1 durable core and is already wired to dispatcher, search indexing, automation delivery, macro fan-out, CSAT survey/template sends, SLA scans, contact exports, Captain document sync, and conversation maintenance jobs. +- `internal/worker/worker.go` has the committed P5.1 durable core and is already wired to dispatcher, search indexing, automation delivery, macro fan-out, CSAT survey/template sends, SLA scans, contact import/export, Captain document sync, and conversation maintenance jobs. - `internal/service/message_delivery_worker.go` and `internal/handler/webhook/incoming_persister_jobs.go` now cover P5.10 SendReplyJob-style outbound delivery plus provider delivery-status/read-receipt jobs. - `internal/handler/webhook/*` and provider services now keep verification/parsing in the HTTP path while `IncomingPersister` moves normalized inbound persistence/dispatch behind durable jobs when the WorkerPool is configured. - `internal/service/captain_*` and `internal/service/copilot_*` now cover Captain document sync, crawl/parser, schedule-sync, response-builder, embedding-update, Copilot response, and Captain conversation response-builder jobs. @@ -1975,6 +1978,7 @@ Tracking table: | 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/service/csat_template_service.go`, channel send services | Resolve-triggered CSAT send is durable; WhatsApp/Twilio template delivery and failure states are fakeable and observable. | Review by `feat(automation): queue macro and csat jobs` and `feat(csat): queue channel templates` | | 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. | Review by `feat(crm): queue contact exports` | +| P5.8b | Queue contact CSV imports through DataImportJob. | `data_import_job.rb`, `DataImport::ContactManager`, `contacts_controller#import` | contact import service and worker boundary | Import API creates a pending data_import and returns immediately, uploaded CSV data survives worker restart, import processing is retryable/observable, and missing files return Chatwoot's exact `File is blank` error. | Review by `feat(crm): queue contact imports` | | 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. | Review by `feat(webhook): queue incoming messages` | | 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. | Review by `feat(messages): queue send replies` and `feat(messages): queue delivery statuses` | | P5.11 | Queue Captain document sync, crawl, response building, embeddings, and Copilot responses. | Captain document/crawl/response/embedding/Copilot/conversation jobs | `internal/service/captain_document_service.go`, `internal/service/copilot_service.go`, `internal/service/captain_conversation_service.go`, Captain/Copilot services | Existing fakeable disabled/failure gates run under durable jobs; document statuses, Copilot message persistence, and Captain conversation replies survive worker restart. | Review by `feat(captain): queue document syncs`, `feat(captain): queue document crawl jobs`, `feat(captain): queue response embedding jobs`, and `feat(captain): queue copilot response jobs` | @@ -2037,7 +2041,7 @@ P5.4 current checkpoint: - The durable job handlers reuse the existing fakeable HTTP webhook and SMTP transcript delivery interfaces, so timeout/retry behavior remains tested at the delivery boundary while worker attempts, errors, retry schedules, completion, and dead-letter state are visible through `background_jobs`. - `AutomationRuleService`, `AutomationRuleListener`, and `MacroService` can carry the WorkerPool through action execution. `Bootstrap` now creates a durable WorkerPool, registers automation delivery jobs, wires the dispatcher with event jobs, and starts the worker in `App.Run`. - Existing no-worker constructors still preserve the previous synchronous fallback for focused tests and development paths that do not start the durable worker yet. -- Remaining Phase 5 work: delayed automation actions if current-reference params require them; macro fan-out, CSAT sends/templates, Meilisearch indexing fan-out, SLA scans, contact export email, provider webhook/offline delivery, Captain/Copilot, conversation maintenance, and analytics rollups/timeseries are in Review. +- Remaining Phase 5 work: delayed automation actions if current-reference params require them; macro fan-out, CSAT sends/templates, Meilisearch indexing fan-out, SLA scans, contact import/export, provider webhook/offline delivery, Captain/Copilot, conversation maintenance, and analytics rollups/timeseries are in Review. P5.4 verification: @@ -2104,6 +2108,24 @@ env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache git diff --check ``` +P5.8b current checkpoint: + +- `contact:import` now mirrors Chatwoot `DataImportJob` for contact CSV imports on the `low` queue. When a WorkerPool is configured, `POST /contacts/import` creates a pending `data_imports` row, stores the uploaded CSV in `import_config` for replay, enqueues the worker job, and returns without creating contacts on the request path. +- Worker replay reloads the `data_imports` row, marks it processing, runs the existing CSV importer, and records completed/failed status plus processed/failed/total counters. Failed replay records the processing error and leaves retry/dead-letter visibility in `background_jobs`. +- Missing `import_file` now returns Chatwoot's exact `422 { error: "File is blank" }` response. No-worker construction still uses the synchronous fallback for focused tests and local paths that do not start the durable worker. +- Remaining P5.8b work: ActiveStorage-style attached import-file metadata is not yet modeled; Go stores durable CSV bytes in `import_config` until a broader file-attachment parity slice requires it. + +P5.8b verification: + +```bash +env GOCACHE=/home/rogee/Projects/gochat/.tmp/go-build GOTMPDIR=/home/rogee/Projects/gochat/.tmp/go-tmp TMPDIR=/home/rogee/Projects/gochat/.tmp/go-tmp go test ./internal/service -run 'ContactService_ImportContacts\|ContactService_ContactImport\|ContactService_ExportContacts_Queues\|ContactService_ContactExport' -count=1 +env GOCACHE=/home/rogee/Projects/gochat/.tmp/go-build GOTMPDIR=/home/rogee/Projects/gochat/.tmp/go-tmp TMPDIR=/home/rogee/Projects/gochat/.tmp/go-tmp go test ./internal/handler/api/v1 -run 'ContactImport\|ContactHandlerCRUDTestSuite/TestImport' -count=1 +env GOCACHE=/home/rogee/Projects/gochat/.tmp/go-build GOTMPDIR=/home/rogee/Projects/gochat/.tmp/go-tmp TMPDIR=/home/rogee/Projects/gochat/.tmp/go-tmp go test ./internal/service -run 'Contact' -count=1 +env GOCACHE=/home/rogee/Projects/gochat/.tmp/go-build GOTMPDIR=/home/rogee/Projects/gochat/.tmp/go-tmp TMPDIR=/home/rogee/Projects/gochat/.tmp/go-tmp go test ./internal/handler/api/v1 -run 'Contact' -count=1 +env GOCACHE=/home/rogee/Projects/gochat/.tmp/go-build GOTMPDIR=/home/rogee/Projects/gochat/.tmp/go-tmp TMPDIR=/home/rogee/Projects/gochat/.tmp/go-tmp go test ./... +git diff --check +``` + P5.9 current checkpoint: - `webhook:incoming_message_persist` now mirrors the persistence/dispatch deferral used by Chatwoot provider event jobs such as `Webhooks::TelegramEventsJob`, `Webhooks::LineEventsJob`, `Webhooks::WhatsappEventsJob`, `Webhooks::TwilioEventsJob`, `Webhooks::FacebookEventsJob`, and `Webhooks::TiktokEventsJob`. @@ -2570,3 +2592,4 @@ Verification milestone gates: - 2026-06-07: P3.52 outgoing messages count builder checkpoint prepared as `feat(reports): align outgoing message counts`; audited Chatwoot v2 reports controller `outgoing_messages_count` and `V2::Reports::OutgoingMessagesCountBuilder`. GoChat invalid or blank `group_by` now returns empty `422`, and label grouping now follows Chatwoot by grouping counts on tag name and resolving the current account label id rather than returning the raw joined tag id. Focused Analytics handler and service tests passed; escalated full `go test ./...` and `git diff --check` passed. No route artifacts change. - 2026-06-07: P3.53 first-response distribution range checkpoint prepared as `feat(reports): align first response distribution range`; audited Chatwoot `V2::Reports::FirstResponseTimeDistributionBuilder` and `DateRangeHelper#range`. GoChat `/reports/first_response_time_distribution` now accepts missing or partial `since/until` as an all-time distribution and only applies the created-at range when both params are present. Focused Analytics handler and service tests passed; escalated full `go test ./...` and `git diff --check` passed. No route artifacts change. - 2026-06-07: Documentation checkpoint prepared as `docs: refresh placeholder audit`; refreshed Phase 6 placeholder evidence and `docs/parity/frontend_smoke_report.md`. The narrow `chatwootParityStub` scan still finds only public webhook nil-handler fallbacks, and the broader placeholder/TODO scan finds no reused-frontend critical account/contact/conversation/message/inbox/widget/public handler placeholder. `scripts/parity_frontend_smoke.sh --check` passed after rerunning with workspace-local `GOCACHE`, `GOTMPDIR`, and `TMPDIR` because `/tmp` was full; `git diff --check` passed. Next default action remains Phase 2/3 drift audit unless the full B12 live stack is available. +- 2026-06-07: P5.8b contact import DataImportJob checkpoint prepared as `feat(crm): queue contact imports`; audited Chatwoot contacts import controller/specs plus `DataImportJob`/contact manager behavior. GoChat contact imports now return the exact missing-file `422 { error: "File is blank" }`, create pending `data_imports` with replayable CSV payloads, enqueue `contact:import` jobs on the low queue when workers are configured, and process imports asynchronously while preserving existing merge, label validation, custom-attribute, status, and counter behavior. Focused contact service and handler tests passed; full `go test ./...` and `git diff --check` passed. No route artifacts change. diff --git a/internal/handler/api/v1/contact_handler.go b/internal/handler/api/v1/contact_handler.go index 242e1077..9d3fb70f 100644 --- a/internal/handler/api/v1/contact_handler.go +++ b/internal/handler/api/v1/contact_handler.go @@ -932,7 +932,7 @@ func (h *ContactHandler) Import(c *gin.Context) { file, _, fileErr = c.Request.FormFile("file") } if fileErr != nil { - c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "failed to import contacts"}) + c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "File is blank"}) return } defer file.Close() diff --git a/internal/handler/api/v1/contact_handler_g3_test.go b/internal/handler/api/v1/contact_handler_g3_test.go index 40449f75..d9b443c8 100644 --- a/internal/handler/api/v1/contact_handler_g3_test.go +++ b/internal/handler/api/v1/contact_handler_g3_test.go @@ -93,7 +93,7 @@ func TestContactImportMissingFile(t *testing.T) { assert.Equal(t, http.StatusUnprocessableEntity, w.Code) var resp map[string]interface{} json.Unmarshal(w.Body.Bytes(), &resp) - assert.Contains(t, resp, "error") + assert.Equal(t, "File is blank", resp["error"]) } func TestContactableInboxesBadAccountID(t *testing.T) { diff --git a/internal/service/contact_import_worker.go b/internal/service/contact_import_worker.go new file mode 100644 index 00000000..c83e7cff --- /dev/null +++ b/internal/service/contact_import_worker.go @@ -0,0 +1,47 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "sync" + + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" +) + +const TaskTypeContactImport = "contact:import" + +type contactImportConfig struct { + CSVBase64 string `json:"csv_base64"` +} + +type contactImportJob struct { + ImportID uint `json:"import_id"` +} + +var contactImportRegistrations sync.Map + +// RegisterContactImportJobs wires Chatwoot DataImportJob contacts import into +// the durable worker. Handlers are registered once per WorkerPool instance. +func RegisterContactImportJobs(wp *worker.WorkerPool, svc *ContactService) { + if wp == nil || svc == nil { + return + } + if _, loaded := contactImportRegistrations.LoadOrStore(wp, struct{}{}); loaded { + return + } + wp.Register(TaskTypeContactImport, svc.performContactImportJob) +} + +func (s *ContactService) performContactImportJob(ctx context.Context, job *model.BackgroundJob) error { + var payload contactImportJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal contact import job: %w", err) + } + if payload.ImportID == 0 { + return fmt.Errorf("invalid contact import job payload: %#v", payload) + } + _, err := s.performContactImport(ctx, payload.ImportID) + return err +} diff --git a/internal/service/contact_service.go b/internal/service/contact_service.go index 93905ccf..5a0de5c6 100644 --- a/internal/service/contact_service.go +++ b/internal/service/contact_service.go @@ -3,6 +3,7 @@ package service import ( "bytes" "context" + "encoding/base64" "encoding/csv" "encoding/json" "errors" @@ -56,6 +57,7 @@ func (s *ContactService) SetContactExportMailer(mailer ContactExportMailer) { func (s *ContactService) SetWorkerPool(wp *worker.WorkerPool) { s.worker = wp RegisterContactExportJobs(wp, s) + RegisterContactImportJobs(wp, s) } func (s *ContactService) indexContact(ctx context.Context, contact *model.Contact) { @@ -995,25 +997,63 @@ func (s *ContactService) ImportContacts(ctx context.Context, accountID, userID u if !s.Ready() { return nil, errors.New("contact service not ready") } + csvData, err := io.ReadAll(r) + if err != nil { + return nil, fmt.Errorf("read import file: %w", err) + } var userIDPtr *uint if userID != 0 { userIDPtr = &userID } - dataImport := &model.DataImport{AccountID: accountID, UserID: userIDPtr, DataType: "contacts", Status: string(model.DataImportStatusPending)} + config, _ := json.Marshal(contactImportConfig{CSVBase64: base64.StdEncoding.EncodeToString(csvData)}) + dataImport := &model.DataImport{AccountID: accountID, UserID: userIDPtr, DataType: "contacts", Status: string(model.DataImportStatusPending), ImportConfig: config} if err := s.repo.DB().WithContext(ctx).Create(dataImport).Error; err != nil { return nil, err } - if err := s.repo.DB().WithContext(ctx).Model(dataImport).Update("status", string(model.DataImportStatusProcessing)).Error; err != nil { + if s.worker != nil { + _, err := s.worker.Enqueue(ctx, TaskTypeContactImport, contactImportJob{ImportID: dataImport.ID}, worker.WithQueue("low"), worker.WithMaxAttempts(3), worker.WithIdempotencyKey(fmt.Sprintf("contact-import:%d", dataImport.ID))) + if err != nil { + s.repo.DB().WithContext(ctx).Model(dataImport).Updates(map[string]any{ + "status": string(model.DataImportStatusFailed), + "processing_errors": err.Error(), + }) + return dataImport, err + } + return dataImport, nil + } + return s.performContactImport(ctx, dataImport.ID) +} + +func (s *ContactService) performContactImport(ctx context.Context, importID uint) (*model.DataImport, error) { + if !s.Ready() { + return nil, errors.New("contact service not ready") + } + var dataImport model.DataImport + if err := s.repo.DB().WithContext(ctx).First(&dataImport, importID).Error; err != nil { + return nil, err + } + if dataImport.Status == string(model.DataImportStatusCompleted) { + return &dataImport, nil + } + var config contactImportConfig + if err := json.Unmarshal(dataImport.ImportConfig, &config); err != nil { + return nil, fmt.Errorf("unmarshal import config: %w", err) + } + csvData, err := base64.StdEncoding.DecodeString(config.CSVBase64) + if err != nil { + return nil, fmt.Errorf("decode import csv: %w", err) + } + if err := s.repo.DB().WithContext(ctx).Model(&dataImport).Update("status", string(model.DataImportStatusProcessing)).Error; err != nil { return nil, err } - result, err := s.ImportCSV(ctx, accountID, r) + result, err := s.ImportCSV(ctx, dataImport.AccountID, bytes.NewReader(csvData)) if err != nil { - s.repo.DB().WithContext(ctx).Model(dataImport).Updates(map[string]any{ + s.repo.DB().WithContext(ctx).Model(&dataImport).Updates(map[string]any{ "status": string(model.DataImportStatusFailed), "processing_errors": err.Error(), }) - return dataImport, err + return &dataImport, err } updates := map[string]any{ "status": string(model.DataImportStatusCompleted), @@ -1021,13 +1061,13 @@ func (s *ContactService) ImportContacts(ctx context.Context, accountID, userID u "failed_records": result.Failed, "total_records": result.Imported + result.Skipped + result.Failed, } - if err := s.repo.DB().WithContext(ctx).Model(dataImport).Updates(updates).Error; err != nil { + if err := s.repo.DB().WithContext(ctx).Model(&dataImport).Updates(updates).Error; err != nil { return nil, err } - if err := s.repo.DB().WithContext(ctx).First(dataImport, dataImport.ID).Error; err != nil { + if err := s.repo.DB().WithContext(ctx).First(&dataImport, dataImport.ID).Error; err != nil { return nil, err } - return dataImport, nil + return &dataImport, nil } // ImportCSV reads contacts from a CSV reader and creates them. diff --git a/internal/service/contact_service_g3_test.go b/internal/service/contact_service_g3_test.go index c384d84a..c66e2573 100644 --- a/internal/service/contact_service_g3_test.go +++ b/internal/service/contact_service_g3_test.go @@ -336,6 +336,62 @@ func TestContactService_ContactExportJobRetriesMissingExport(t *testing.T) { assert.NotEmpty(t, job.LastError) } +func TestContactService_ImportContacts_QueuesDurableDataImport(t *testing.T) { + db, _, svc := setupContactService(t) + account := createTestAccount(t, db) + userID := uint(42) + require.NoError(t, db.Create(&model.Tag{AccountID: account.ID, Name: "vip"}).Error) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 7, 9, 0, 0, 0, time.UTC) })) + svc.SetWorkerPool(wp) + + dataImport, err := svc.ImportContacts(context.Background(), account.ID, userID, strings.NewReader("name,email,labels\nQueued,queued@test.com,vip\n")) + require.NoError(t, err) + assert.Equal(t, "contacts", dataImport.DataType) + assert.Equal(t, string(model.DataImportStatusPending), dataImport.Status) + assert.NotEmpty(t, dataImport.ImportConfig) + require.NotNil(t, dataImport.UserID) + assert.Equal(t, userID, *dataImport.UserID) + + var contactCount int64 + require.NoError(t, db.Model(&model.Contact{}).Where("account_id = ? AND email = ?", account.ID, "queued@test.com").Count(&contactCount).Error) + assert.Equal(t, int64(0), contactCount) + + var jobCount int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ? AND queue = ? AND status = ?", TaskTypeContactImport, "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.DataImport + require.NoError(t, db.First(&completed, dataImport.ID).Error) + assert.Equal(t, string(model.DataImportStatusCompleted), completed.Status) + assert.Equal(t, 1, completed.TotalRecords) + assert.Equal(t, 1, completed.ProcessedRecords) + + require.NoError(t, db.Model(&model.Contact{}).Where("account_id = ? AND email = ?", account.ID, "queued@test.com").Count(&contactCount).Error) + assert.Equal(t, int64(1), contactCount) +} + +func TestContactService_ContactImportJobRetriesMissingImport(t *testing.T) { + db, _, svc := setupContactService(t) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 7, 9, 15, 0, 0, time.UTC) }), worker.WithBackoff(func(attempt int) time.Duration { return time.Minute })) + svc.SetWorkerPool(wp) + + _, err := wp.Enqueue(context.Background(), TaskTypeContactImport, contactImportJob{ImportID: 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 import to retry, processed=%v err=%v", processed, err) + } + + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeContactImport).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)