Migrating topic → CustomerIssue assignment to API2 BullMQ

Why

Post-embedding CustomerIssue assignment used to bounce between Rails and API2:

Rails CompleteConversationSyncJob
  → API2 conversation-topic-embedding
    → POST /internal/conversations/assign_customer_issue_from_topic (Rails)
      → Rails GoodJob AssignConversationToCustomerIssueFromTopicJob
        → POST /internal/customer_issue_embeddings/match (API2)
          → Rails Conversation#assign + PatchConversationTopicTaggingJob → API2

Every untagged synced conversation crossed the Rails↔API2 boundary four times. During the Kuda (org 25) backfill this accumulated ~78k unfinished GoodJob rows for ~67k conversations (~13.6% duplicate conversation ids).

After

Assignment runs entirely inside API2 on its own BullMQ queue:

API2 conversation-topic-embedding (indexes topic doc)
  → enqueue topic-customer-issue-assignment (BullMQ, deterministic job id)
    → runTopicCustomerIssueAssignmentJob:
        field-mapping precedence → Voyage ANN nearest-match
        race-safe UPDATE ... WHERE customer_issue_id IS NULL
        patch Turbopuffer conversation topic doc

Key properties (unchanged from Rails semantics):

• Only existing CustomerIssue rows are used. No issue is ever created, no LLM is called.
• Field-mapped issues take precedence over the embedding nearest-match. Finding a configured field value stops the fallback even if the issue row is missing only when the value does not map; a present-but-already-assigned conversation is left untouched.
• The write is tenant-scoped and race-safe: customer_issue_id is set only while it is still NULL for that (id, organization_id), so a concurrent or newer Rails write is never clobbered. Turbopuffer is patched only when this job wins.
• BullMQ job id is topic_customer_issue_assignment_<org>_<conversation>_<sourceVersion>, where sourceVersion is the embedding source version (direct path) or the conversation’s last_synced_at/updated_at (Rails bridge). Repeated syncs and duplicate backlog rows for the same version collapse onto one job, but a later sync with changed topic text produces a new job id and is not suppressed by a retained terminal (completed/no_match/failed) job.
• Retries are exponential (attempts: 5, 30s base backoff). Concurrency is bounded at two levels: per-process concurrency (TOPIC_CUSTOMER_ISSUE_ASSIGNMENT_WORKER_CONCURRENCY, default 4) and a regional/global cap via BullMQ setGlobalConcurrency (TOPIC_CUSTOMER_ISSUE_ASSIGNMENT_GLOBAL_CONCURRENCY, default 4) so the total across all worker replicas stays bounded and does not hammer Turbopuffer. Sentry queue instrumentation is inherited from SentryQueue/bindWorkerEvents.
• Turbopuffer patching is retry-safe: if the DB assignment commits but the patch throws, BullMQ retries and the already-tagged branch idempotently re-patches the topic document with the current customer_issue_id instead of leaving it stale.

Rails producers enqueue API2 directly (ENG-2252)

No Rails producer routes through a GoodJob worker anymore. The incident showed that even a thin HTTP bridge job consumed tens of thousands of shared latency_30s slots, so every producer now calls Conversation#enqueue_customer_issue_assignment_from_topic!, which POSTs straight to POST /internal/conversation_topic_customer_issue_assignments/enqueue and raises Conversation::Api2CustomerIssueAssignmentEnqueueError on delivery failure. Producers:

• Internal::ConversationTopicCustomerIssueAssignmentsController#create — the API2 callback path. Enqueues inline; a delivery failure returns 502 so the caller can retry.
• Maintenance::AssignUntaggedConversationsToCustomerIssuesFromTopicTask and scripts/enqueue_untagged_topic_customer_issue_assigns.rb — backfills. Enqueue inline per conversation; a per-record delivery failure is logged and skipped so one blip does not abort the run. The deterministic API2 job id makes re-runs idempotent.

The deterministic API2 job id dedupes the ~13.6% duplicate conversation ids for free, exactly as the old bridge did.

Legacy GoodJob backlog drain shim

AssignConversationToCustomerIssueFromTopicJob is kept — but nothing enqueues it. It is now a minimal compatibility drain shim whose perform just calls Conversation#enqueue_customer_issue_assignment_from_topic!, so the existing ~78k GoodJob rows queued before ENG-2252 (queue latency_30s) still deserialize and drain by handing off to API2. Delivery failures are retried by GoodJob (retry_on, attempts: 10).

Deletion condition: remove the shim once no AssignConversationToCustomerIssueFromTopicJob rows remain in the production GoodJob tables (the latency_30s backlog is empty). Because nothing enqueues it, an empty backlog is a sufficient signal that it is safe to delete.

Internal::ConversationTopicCustomerIssueAssignmentsController and the Rails Conversation#assign_customer_issue_from_topic_embedding model methods are left in place for backward compatibility / rollback during rollout.

Rollout sequence

1. Deploy API2 (new queue, worker, enqueue endpoint) and Rails (bridge job) together. The new embedding completions immediately enqueue API2 assignment; the Rails endpoint keeps working for any in-flight callers.
2. Confirm the topic-customer-issue-assignment worker is processing and Sentry shows healthy completion/fail rates.
3. Let the existing GoodJob backlog drain: as each AssignConversationToCustomerIssueFromTopicJob runs it enqueues (idempotently) into API2. Do not delete production GoodJob rows; let them complete or age out normally.
4. Monitor conversations.customer_issue_id fill rate and Turbopuffer is_untagged counts for org 25 to confirm parity with the old path.

Follow-up removal (after rollout is confirmed)

Once no traffic depends on the legacy path:

• Remove POST /internal/conversations/assign_customer_issue_from_topic (Internal::ConversationTopicCustomerIssueAssignmentsController).
• Delete the AssignConversationToCustomerIssueFromTopicJob drain shim once the latency_30s GoodJob backlog for it is empty (nothing enqueues it anymore).
• Remove the now-unused Rails Conversation assignment methods (assign_customer_issue_from_topic_embedding, apply_existing_field_mapped_customer_issue, assign_from_nearest_customer_issues_for_topic_text, and the POST /internal/customer_issue_embeddings/match caller) plus the PatchConversationTopicTaggingJob if it has no other callers.