Phase1

apps/worker/src/config.ts

@@ -0,0 +1,38 @@
+import { z } from 'zod';
+
+export const notificationEmailJobSchema = z.object({
+  type: z.enum(['email_verification', 'password_reset']),
+  to: z.string().email(),
+  subject: z.string(),
+  template: z.string(),
+  data: z.record(z.unknown()),
+});
+
+export type NotificationEmailJob = z.infer<typeof notificationEmailJobSchema>;
+
+const smtpConfigSchema = z.object({
+  SMTP_HOST: z.string().min(1).default('localhost'),
+  SMTP_PORT: z.coerce.number().int().default(1025),
+  SMTP_USER: z.string().optional(),
+  SMTP_PASSWORD: z.string().optional(),
+  SMTP_FROM: z.string().default('noreply@localhost'),
+  SMTP_TLS: z
+    .string()
+    .optional()
+    .transform((v) => v === 'true' || v === '1'),
+  APP_NAME: z.string().default('HexaHost GameCloud'),
+});
+
+export type SmtpConfig = z.infer<typeof smtpConfigSchema>;
+
+export function getSmtpConfig(): SmtpConfig {
+  return smtpConfigSchema.parse({
+    SMTP_HOST: process.env['SMTP_HOST'],
+    SMTP_PORT: process.env['SMTP_PORT'],
+    SMTP_USER: process.env['SMTP_USER'],
+    SMTP_PASSWORD: process.env['SMTP_PASSWORD'],
+    SMTP_FROM: process.env['SMTP_FROM'],
+    SMTP_TLS: process.env['SMTP_TLS'],
+    APP_NAME: process.env['APP_NAME'],
+  });
+}

apps/worker/src/email.ts

@@ -0,0 +1,45 @@
+import nodemailer from 'nodemailer';
+import type { Transporter } from 'nodemailer';
+
+import { getSmtpConfig, type SmtpConfig } from './config';
+
+let transporter: Transporter | null = null;
+
+function getTransporter(config: SmtpConfig): Transporter {
+  if (!transporter) {
+    transporter = nodemailer.createTransport({
+      host: config.SMTP_HOST,
+      port: config.SMTP_PORT,
+      secure: config.SMTP_TLS,
+      auth:
+        config.SMTP_USER && config.SMTP_PASSWORD
+          ? { user: config.SMTP_USER, pass: config.SMTP_PASSWORD }
+          : undefined,
+    });
+  }
+  return transporter;
+}
+
+export interface SendEmailPayload {
+  to: string;
+  subject: string;
+  html: string;
+  text?: string;
+}
+
+export async function sendEmail(payload: SendEmailPayload): Promise<void> {
+  const config = getSmtpConfig();
+  const transport = getTransporter(config);
+
+  await transport.sendMail({
+    from: config.SMTP_FROM,
+    to: payload.to,
+    subject: payload.subject,
+    html: payload.html,
+    text: payload.text ?? stripHtml(payload.html),
+  });
+}
+
+function stripHtml(html: string): string {
+  return html.replace(/<[^>]+>/g, ' ').replace(/\s+/g, ' ').trim();
+}

apps/worker/src/handlers/notifications.ts

@@ -0,0 +1,33 @@
+import { Job } from 'bullmq';
+
+import { getSmtpConfig, notificationEmailJobSchema } from '../config';
+import { sendEmail } from '../email';
+import { logger } from '../logger';
+
+export async function processNotificationJob(job: Job): Promise<{ sent: boolean }> {
+  if (job.name !== 'send-email') {
+    logger.warn({ jobName: job.name }, 'Unknown notification job');
+    return { sent: false };
+  }
+
+  const payload = notificationEmailJobSchema.parse(job.data);
+  const config = getSmtpConfig();
+
+  let html: string;
+  if (payload.type === 'email_verification') {
+    const verifyUrl = String(payload.data['verifyUrl'] ?? '');
+    html = `<p>Welcome to ${config.APP_NAME}!</p><p><a href="${verifyUrl}">Verify your email address</a></p><p>This link expires in 24 hours.</p>`;
+  } else {
+    const resetUrl = String(payload.data['resetUrl'] ?? '');
+    html = `<p>You requested a password reset for ${config.APP_NAME}.</p><p><a href="${resetUrl}">Set a new password</a></p><p>If you did not request this, ignore this email.</p>`;
+  }
+
+  await sendEmail({
+    to: payload.to,
+    subject: payload.subject,
+    html,
+  });
+
+  logger.info({ to: payload.to, type: payload.type, jobId: job.id }, 'Email sent');
+  return { sent: true };
+}

apps/worker/src/index.ts

@@ -3,6 +3,7 @@ import { Job, Worker, type ConnectionOptions } from 'bullmq';
 import { validateConfig } from '@hexahost/config';
 import { prisma } from '@hexahost/database';
 
+import { processNotificationJob } from './handlers/notifications';
 import { startHealthServer } from './health';
 import { logger } from './logger';
 import { WORKER_QUEUES, type WorkerQueueName } from './queues';
@@ -19,6 +20,10 @@ function createQueueWorker(
         'Processing job',
       );
 
+      if (queueName === 'notifications') {
+        return processNotificationJob(job);
+      }
+
       return { acknowledged: true, queue: queueName };
     },
     { connection },