feat(queue): implement real email transport in the email processor

docs/email-notifications.md

@@ -1,6 +1,46 @@
-# Outbound Notification Subsystem
+# Outbound Email Notifications
 
-This guide documents the outbound notification subsystem so contributors can
+This guide covers both synchronous notification dispatch and the asynchronous
+queue-based email processor.
+
+## Queue email processor
+
+Queued `email-notification` jobs are handled by
+[`src/queue/processors/email-processor.ts`](../src/queue/processors/email-processor.ts)
+using the pluggable transport in
+[`src/queue/processors/email.transport.ts`](../src/queue/processors/email.transport.ts).
+
+### Flow
+
+1. Validate the recipient with a strict single-address check (CR/LF, multi-recipient
+   separators, and non-RFC shapes are rejected).
+2. Require non-empty `subject` and `body`.
+3. Guard `to` / `subject` / `body` against SMTP header injection.
+4. Dispatch through the configured {@link EmailTransport}; provider failures throw
+   so BullMQ retries the job instead of marking it successful.
+5. Return `{ success, message, data: { emailId } }` on success.
+
+### Transport selection
+
+`resolveEmailTransport()` reads validated config from `src/config/env.schema.ts`:
+
+| `EMAIL_PROVIDER` | Transport class           | Required config                               |
+| ---------------- | ------------------------- | --------------------------------------------- |
+| `console`        | `ConsoleEmailTransport`   | — (default in development/test)               |
+| `smtp`           | `SmtpEmailTransport`      | `SMTP_HOST`, `SMTP_PORT`, `SMTP_FROM`         |
+| `ses`            | `SesEmailTransport`       | `SMTP_FROM`, `AWS_REGION`, AWS credentials    |
+| `sendgrid`       | `SendGridEmailTransport`  | `SMTP_FROM`, `SENDGRID_API_KEY`               |
+
+Set `EMAIL_SEND_TIMEOUT_MS` (default `10000`) to bound provider calls.
+
+Tests inject a mock via `setEmailTransportOverride()`; production code must not
+log full recipient addresses or bodies at info level.
+
+---
+
+## Notification service subsystem
+
+The following section documents the outbound notification subsystem so contributors can
 extend it safely. It reflects the code as implemented in:
 
 - Orchestration: [`src/services/notification.service.ts`](../src/services/notification.service.ts)

src/config/env.schema.ts

@@ -164,6 +164,37 @@ export const envSchema = z.object({
 
   REPUTATION_SCORE_ALGORITHM_VERSION: z.string()
     .default('exp-decay-v1'),
+
+  // Email transport (queue processor + notification service)
+  EMAIL_PROVIDER: z.enum(['console', 'smtp', 'ses', 'sendgrid'])
+    .default('console'),
+
+  EMAIL_SEND_TIMEOUT_MS: z.string()
+    .default('10000')
+    .transform((val) => parseInt(val, 10))
+    .pipe(z.number().int().min(1000).max(120_000)),
+
+  SMTP_HOST: z.string().optional(),
+  SMTP_PORT: z.string()
+    .optional()
+    .transform((val) => val === undefined ? undefined : parseInt(val, 10))
+    .pipe(z.number().int().min(1).max(65535).optional()),
+  SMTP_USER: z.string().optional(),
+  SMTP_PASSWORD: z.string().optional(),
+  SMTP_FROM: z.string()
+    .optional()
+    .refine((val) => val === undefined || !/[\r\n]/.test(val), {
+      message: 'SMTP_FROM must not contain CR/LF characters',
+    }),
+  SMTP_SECURE: z.string()
+    .optional()
+    .transform((val) => val === undefined ? undefined : val === 'true'),
+
+  AWS_ACCESS_KEY_ID: z.string().optional(),
+  AWS_SECRET_ACCESS_KEY: z.string().optional(),
+  AWS_REGION: z.string().optional(),
+
+  SENDGRID_API_KEY: z.string().optional(),
 }).superRefine((obj, ctx) => {
   if (obj.NODE_ENV !== 'test') {
     if (!obj.JWT_SECRET) {

src/queue/processors/email-processor.test.ts

@@ -6,6 +6,10 @@
  */
 
 import { processEmailNotification, generateEmailId } from './email-processor';
+import {
+  EmailTransport,
+  setEmailTransportOverride,
+} from './email.transport';
 import { EmailNotificationPayload } from '../types';
 import { setWriteRecordImpl, LogRecord } from '../../logger';
 
@@ -48,6 +52,10 @@ describe('generateEmailId', () => {
 // ── processEmailNotification ──────────────────────────────────────────────────
 
 describe('processEmailNotification', () => {
+  afterEach(() => {
+    setEmailTransportOverride(undefined);
+  });
+
   describe('happy path', () => {
     it('returns success with a crypto-strong emailId', async () => {
       const result = await processEmailNotification(BASE_PAYLOAD);
@@ -84,6 +92,24 @@ describe('processEmailNotification', () => {
       ).rejects.toThrow('Invalid email address');
     });
 
+    it('throws on header-injection in recipient', async () => {
+      await expect(
+        processEmailNotification({
+          ...BASE_PAYLOAD,
+          to: 'user@example.com\r\nBcc: evil@example.com',
+        }),
+      ).rejects.toThrow('Invalid email address');
+    });
+
+    it('throws on header-injection in subject', async () => {
+      await expect(
+        processEmailNotification({
+          ...BASE_PAYLOAD,
+          subject: 'Hello\r\nBcc: evil@example.com',
+        }),
+      ).rejects.toThrow('header injection');
+    });
+
     it('throws on missing subject', async () => {
       await expect(
         processEmailNotification({ ...BASE_PAYLOAD, subject: '' }),
@@ -170,6 +196,29 @@ describe('processEmailNotification', () => {
       }
     });
 
+    it('propagates transport failures so the job can be retried', async () => {
+      const failingTransport: EmailTransport = {
+        send: async () => {
+          throw new Error('Provider unavailable');
+        },
+      };
+      setEmailTransportOverride(failingTransport);
+
+      await expect(processEmailNotification(BASE_PAYLOAD)).rejects.toThrow(
+        'Provider unavailable',
+      );
+    });
+
+    it('invokes the injected transport on success', async () => {
+      const send = jest.fn().mockResolvedValue({ providerMessageId: 'msg-1' });
+      setEmailTransportOverride({ send });
+
+      const result = await processEmailNotification(BASE_PAYLOAD);
+
+      expect(send).toHaveBeenCalledTimes(1);
+      expect(result.success).toBe(true);
+    });
+
     it('includes emailId in the delivery log record', async () => {
       const { records, restore } = captureRecords();
       let result: Awaited<ReturnType<typeof processEmailNotification>>;

src/queue/processors/email-processor.ts

@@ -7,6 +7,11 @@
 
 import { EmailNotificationPayload, JobResult } from '../types';
 import { createLogger } from '../../logger';
+import {
+  assertSafeEmailHeaders,
+  isValidRecipientEmail,
+  resolveEmailTransport,
+} from './email.transport';
 
 /**
  * Generate a cryptographically-strong unique tracking ID for an outbound email.
@@ -24,9 +29,13 @@ export function generateEmailId(): string {
 /**
  * Process email notification job
  *
+ * Validates the recipient, guards against header injection, dispatches through
+ * the configured {@link EmailTransport}, and surfaces provider failures so the
+ * queue manager can retry the job.
+ *
  * @param payload - Email notification data
  * @returns Job result with success status
- * @throws Error if email validation fails
+ * @throws Error if validation or delivery fails
  */
 export async function processEmailNotification(
   payload: EmailNotificationPayload,
@@ -37,29 +46,40 @@ export async function processEmailNotification(
     ...(payload.requestId && { requestId: payload.requestId }),
   });
 
-  // Validate email format
-  const emailRegex = /^[^\s@]+@[^\s@]+\.[^\s@]+$/;
-  if (!emailRegex.test(payload.to)) {
-    // Do not echo the address at info/error level — it is PII
+  if (!isValidRecipientEmail(payload.to)) {
     log.warn('Email validation failed: invalid address format');
     throw new Error(`Invalid email address: ${payload.to}`);
   }
 
-  // Validate required fields
   if (!payload.subject || !payload.body) {
     log.warn('Email validation failed: missing subject or body');
     throw new Error('Email subject and body are required');
   }
 
-  log.info('Sending email notification', {
+  assertSafeEmailHeaders({
+    to: payload.to,
     subject: payload.subject,
+    body: payload.body,
     templateId: payload.templateId,
-    // recipient address is omitted — it is in SENSITIVE_PARAMS
   });
 
-  await simulateEmailSend(payload, log);
+  log.info('Sending email notification', {
+    subject: payload.subject,
+    templateId: payload.templateId,
+  });
 
   const emailId = generateEmailId();
+  const transport = resolveEmailTransport();
+
+  await transport.send(
+    {
+      to: payload.to,
+      subject: payload.subject,
+      body: payload.body,
+      templateId: payload.templateId,
+    },
+    log,
+  );
 
   log.info('Email notification delivered', { emailId, subject: payload.subject });
 
@@ -69,19 +89,3 @@ export async function processEmailNotification(
     data: { emailId },
   };
 }
-
-/**
- * Simulate email sending with artificial delay.
- * Replace with actual email service API call in production.
- */
-async function simulateEmailSend(
-  payload: EmailNotificationPayload,
-  log: ReturnType<typeof createLogger>,
-): Promise<void> {
-  return new Promise((resolve) => {
-    setTimeout(() => {
-      log.debug('Email delivery simulation complete', { subject: payload.subject });
-      resolve();
-    }, 100);
-  });
-}