[SES-PHASE-1-4] - PLU-744: add consumer for sqs events

[m0nggh]

TL;DR

Adds an SQS-backed consumer for processing SES email events (bounces, complaints, etc.) directly in the worker process using sqs-consumer.

What changed?

• Added @aws-sdk/client-sqs and sqs-consumer as dependencies.
• Introduced ses-consumer.ts, which creates an SQS long-polling consumer via Consumer.create with a batch size of 10 and a 20-second wait time. Messages are parsed via the existing parseSqsMessage helper and dispatched to processSesEvent.
• Returning the message from handleMessage signals sqs-consumer to delete (ack) it; throwing propagates the error so SQS redelivers the message according to the queue's MaxReceiveCount.
• Malformed/unparseable messages are treated as poison messages: they are logged with a body preview and immediately acked to prevent indefinite redelivery until a DLQ is in place.
• Structured error logging is wired to the consumer's error and processing_error events.
• Graceful shutdown is handled via a SIGTERM listener that calls consumer.stop() and awaits the stopped event before logging completion. pollingCompleteWaitTimeMs caps the internal wait, so no external timeout race is needed.
• startSesConsumer is idempotent — repeated calls after the consumer is already running are no-ops.
• Added sqsQueueUrl to the app config, sourced from the SQS_QUEUE_URL environment variable. If unset, the consumer does not start and logs an informational message.
• Registered startSesConsumer in the main worker entrypoint (worker.ts).

How to test?

1. Set SQS_QUEUE_URL to a valid AWS SQS queue URL in your environment.
2. Ensure AWS credentials with SQS read and delete permissions are available (done)
3. Start the worker process and confirm the log line SES consumer started appears.
4. Send a valid email and it should go to the SQS queue and verify processSesEvent is invoked and the message is deleted from the queue.
5. Send to a bounced email address: bounce@simulator.amazonses.com and it should go to the SQS queue and verify processSesEvent is invoked and the message is deleted from the queue.
   1. The entry should be inserted into the DB table
6. Omit SQS_QUEUE_URL and confirm the consumer does not start and logs SQS_QUEUE_URL not set.

Why make this change?

SES delivers email event notifications (bounces, complaints, deliveries) via SNS→SQS. This consumer reliably ingests those events directly in the worker process with structured error handling and graceful shutdown, rather than requiring an additional BullMQ queue layer for this use case.

[m0nggh]

• [SES-PHASE-1-9] - PLU-744: send attachments with ses #1636
• [SES-PHASE-1-8] - PLU-744: add bounce rate tracking (exploration) #1634
• [SES-PHASE-1-7] - PLU-744: add env vars for ses and sqs #1633
• [SES-PHASE-1-6] - PLU-744: add admin whitelist endpoint #1632
• [SES-PHASE-1-5] - PLU-744: add suppression check for ses send email #1631
• [SES-PHASE-1-4] - PLU-744: add consumer for sqs events #1630 👈 (View in Graphite)
• [SES-PHASE-1-3] - PLU-744: add ses event parser #1629
• [SES-PHASE-1-2] - PLU-744: add migration and email suppression model #1628
• [SES-PHASE-1-1] - PLU-744: add SES call and LD flag #1627
• feat/ses/trunk

This stack of pull requests is managed by Graphite. Learn more about stacking.

[linear]

PLU-744

[ogp-weeloong]

no need extra queue here

[ogp-weeloong]

generally OK.

[Copilot]

Pull request overview

Adds an SQS-backed SES event consumer to the backend worker process, enabling the worker to long-poll an SES SNS→SQS queue and dispatch parsed SES events to the existing processSesEvent handler with structured logging and graceful shutdown behavior.

Changes:

• Start an SES SQS consumer from the worker entrypoint when SQS_QUEUE_URL is configured.
• Add ses-consumer.ts implementing the sqs-consumer wiring, poison-message handling, and SIGTERM shutdown.
• Add config + unit tests for the new consumer and introduce required SQS dependencies.

Reviewed changes

Copilot reviewed 5 out of 6 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
packages/backend/src/worker.ts	Starts the SES SQS consumer when the worker boots.
packages/backend/src/helpers/ses-consumer.ts	Implements SQS long-polling consumer, parsing, dispatch, logging, and SIGTERM shutdown.
packages/backend/src/helpers/tests/ses-consumer.test.ts	Unit tests for consumer start/stop/idempotency, poison-message handling, and SIGTERM flow.
packages/backend/src/config/app.ts	Adds ses.sqsQueueUrl sourced from SQS_QUEUE_URL.
packages/backend/package.json	Adds @aws-sdk/client-sqs and sqs-consumer dependencies.
package-lock.json	Locks new dependency tree for SQS consumer support.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.