email.processor.ts

email.processor.tsapps/worker/src/processors/email.processor.tsraw ← processors10,877 B← Back to the app
1/**
2 * apps/worker — email queue processor (outbound mail).
3 *
4 * Job data: {to, subject, html?, text?, template?, vars?}
5 *
6 * Dev-safe default: when SMTP_HOST is empty the job is a no-op that logs
7 * "[email] (no SMTP configured, skipping)" and completes — local dev must
8 * never require a mail server. Documented in README.
9 *
10 * Templates: welcome, invite, mention, password-reset (6-digit code),
11 * digest. A template + vars is rendered to {subject, html, text} before
12 * sending; `{{var}}` interpolation over the vars object.
13 *
14 * Transport: the dependency-free SMTP client in lib/smtp.ts (EHLO,
15 * optional STARTTLS, AUTH LOGIN, dot-stuffed DATA). SMTP failures throw →
16 * BullMQ retries (8 × exponential).
17 */
18import { Worker, type Job } from "bullmq";
19import { withJobOtel } from "../lib/job-otel.js";
20import type { Redis } from "ioredis";
21import { z } from "zod";
22import type { PrismaClient } from "@prisma/client";
23import { QUEUES } from "../queues.js";
24import { getLogger } from "../logger.js";
25import type { WorkerEnv } from "../config.js";
26import { smtpSend } from "../lib/smtp.js";
27 
28const log = getLogger("email");
29 
30export const emailJobSchema = z.object({
31  to: z.string().email(),
32  cc: z.array(z.string().email()).optional(),
33  subject: z.string().min(1),
34  html: z.string().optional(),
35  text: z.string().optional(),
36  template: z
37    .enum(["welcome", "invite", "mention", "password-reset", "digest", "automation"])
38    .optional(),
39  vars: z.record(z.string(), z.unknown()).optional(),
40  /**
41   * TASK-0094 R6 — the Notification row(s) this email belongs to. An INSTANT
42   * notification email carries `[notificationId]`; a DIGEST email carries the
43   * collected ids. On a successful send the worker stamps every row here with
44   * `emailSentAt` (the idempotency flag — a job redelivered after a partial
45   * SMTP failure must not double-send; the job-key dedup guards the enqueue
46   * race, this flag guards the redelivery). Backward-compatible: existing
47   * enqueue paths (verify / invite / admin alerts) omit it.
48   */
49  notificationIds: z.array(z.string().min(1)).optional()
50});
51export type EmailJobData = z.infer<typeof emailJobSchema>;
52 
53// ---------------------------------------------------------------------------
54// Templates
55// ---------------------------------------------------------------------------
56 
57function interpolate(s: string, vars: Record<string, unknown>): string {
58  return s.replace(/\{\{\s*([\w.]+)\s*\}\}/g, (_m, path: string) => {
59    const v = path.split(".").reduce<unknown>(
60      (acc, k) =>
61        acc && typeof acc === "object" ? (acc as Record<string, unknown>)[k] : undefined,
62      vars
63    );
64    if (v === undefined || v === null) return "";
65    if (typeof v === "object") return JSON.stringify(v);
66    return String(v);
67  });
68}
69 
70interface TemplateRender {
71  subject: string;
72  html: string;
73  text: string;
74}
75 
76const TEMPLATES: Record<
77  string,
78  (vars: Record<string, unknown>) => TemplateRender
79> = {
80  welcome: (v) => {
81    const subject = `Welcome to ${interpolate("{{workspace}}", v)}`;
82    const html =
83      `<h1>Welcome to ${esc(interpolate("{{workspace}}", v))}!</h1>` +
84      `<p>Hi ${esc(interpolate("{{name}}", v))},</p>` +
85      `<p>Your workspace is ready. <a href="{{url}}">Open your workspace</a>.</p>`;
86    const text = `Welcome to ${interpolate("{{workspace}}", v)}!\nHi ${interpolate("{{name}}", v)},\nYour workspace is ready: ${interpolate("{{url}}", v)}`;
87    return page(subject, html, text, v);
88  },
89  invite: (v) => {
90    const subject = `${interpolate("{{inviter}}", v)} invited you to ${interpolate("{{workspace}}", v)}`;
91    const html =
92      `<p><strong>${esc(interpolate("{{inviter}}", v))}</strong> invited you to join ` +
93      `<strong>${esc(interpolate("{{workspace}}", v))}</strong> as ${esc(interpolate("{{role}}", v))}.</p>` +
94      `<p><a href="{{url}}">Accept invitation</a> (valid for 7 days).</p>`;
95    const text = `${interpolate("{{inviter}}", v)} invited you to ${interpolate("{{workspace}}", v)} as ${interpolate("{{role}}", v)}.\nAccept: ${interpolate("{{url}}", v)} (valid for 7 days)`;
96    return page(subject, html, text, v);
97  },
98  mention: (v) => {
99    const subject = `${interpolate("{{mentioner}}", v)} mentioned you: ${interpolate("{{context}}", v)}`;
100    const html =
101      `<p><strong>${esc(interpolate("{{mentioner}}", v))}</strong> mentioned you in ` +
102      `<em>${esc(interpolate("{{context}}", v))}</em>:</p>` +
103      `<blockquote>${esc(interpolate("{{quote}}", v))}</blockquote>` +
104      `<p><a href="{{url}}">View in workspace</a></p>`;
105    const text = `${interpolate("{{mentioner}}", v)} mentioned you in ${interpolate("{{context}}", v)}:\n"${interpolate("{{quote}}", v)}"\n${interpolate("{{url}}", v)}`;
106    return page(subject, html, text, v);
107  },
108  "password-reset": (v) => {
109    const subject = "Your Factory password reset code";
110    const html =
111      `<p>Your password reset code is:</p>` +
112      `<p style="font-size:28px;font-weight:bold;letter-spacing:4px">${esc(interpolate("{{code}}", v))}</p>` +
113      `<p>It expires in 30 minutes. If you did not request this, ignore this email.</p>`;
114    const text = `Your password reset code is ${interpolate("{{code}}", v)}. It expires in 30 minutes.`;
115    return page(subject, html, text, v);
116  },
117  // TASK-0112 SEND_EMAIL: the executor sends fully-rendered html/text (markdown
118  // already converted + interpolated); this template is a pass-through so the
119  // `automation` label is a valid template without imposing its own layout.
120  automation: (v) => {
121    const subject = typeof v.subject === "string" && v.subject ? v.subject : "Message from Factory";
122    const html = typeof v.html === "string" ? v.html : `<pre>${esc(String(v.text ?? ""))}</pre>`;
123    const text = typeof v.text === "string" ? v.text : "";
124    return page(subject, html, text, v);
125  },
126  digest: (v) => {
127    // TASK-0094 R6: the hourly digest. Subject is FIXED ("Your Factory digest" —
128    // the spec + integration test assert it); the body lists the collected
129    // items (PLAIN-TEXT titles passed via vars.items, escaped here) and links
130    // to the inbox (the "n more" path).
131    const items = Array.isArray(v.items) ? (v.items as string[]) : [];
132    const html =
133      `<h2>Your Factory digest</h2>` +
134      `<ul>${items.map((i) => `<li>${esc(i)}</li>`).join("")}</ul>` +
135      `<p><a href="{{url}}">Open your inbox</a></p>`;
136    const text = `Your Factory digest:\n${items.map((i) => `- ${i}`).join("\n")}\n${interpolate("{{url}}", v)}`;
137    return page("Your Factory digest", html, text, v);
138  }
139};
140 
141/** Build a template render from a subject + inner html + plain text. */
142function page(subject: string, html: string, text: string, vars: Record<string, unknown>): TemplateRender {
143  // Templates interpolate most tokens inline (with escaping); pass the whole
144  // block through once more so link hrefs like {{url}} resolve.
145  const fullHtml = interpolate(html, vars);
146  return {
147    subject,
148    html:
149      `<!doctype html><html><body style="font-family:system-ui,sans-serif;color:#191919;max-width:560px;margin:0 auto;padding:24px">` +
150      fullHtml +
151      `<hr style="border:none;border-top:1px solid #e0e0e0;margin-top:32px">` +
152      `<p style="font-size:12px;color:#999">AI Software Factory — you are receiving this because of an activity in your workspace.</p>` +
153      `</body></html>`,
154    text
155  };
156}
157 
158function esc(s: string): string {
159  return s.replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;");
160}
161 
162/** Resolve a job into final {subject, html, text} (template or raw). */
163export function renderEmail(
164  data: EmailJobData
165): { subject: string; html: string; text: string } {
166  if (data.template) {
167    const tpl = TEMPLATES[data.template];
168    if (!tpl) throw new Error(`unknown email template: ${data.template}`);
169    const vars = data.vars ?? {};
170    const r = tpl(vars);
171    return {
172      subject: data.subject || r.subject,
173      html: r.html,
174      text: r.text
175    };
176  }
177  return {
178    subject: data.subject,
179    html: data.html ?? `<pre>${esc(data.text ?? data.subject)}</pre>`,
180    text: data.text ?? ""
181  };
182}
183 
184// ---------------------------------------------------------------------------
185// Worker
186// ---------------------------------------------------------------------------
187 
188/**
189 * TASK-0094 R6 — mark Notification row(s) as emailed. Best-effort: the mail is
190 * already delivered when this runs, so a stamp failure is logged, not thrown
191 * (a throw here would retry an already-sent job — the opposite of what the
192 * idempotency flag exists to prevent).
193 */
194export async function stampEmailSent(
195  prisma: PrismaClient,
196  notificationIds: string[]
197): Promise<void> {
198  try {
199    await prisma.notification.updateMany({
200      where: { id: { in: notificationIds }, emailSentAt: null },
201      data: { emailSentAt: new Date() }
202    });
203  } catch (err) {
204    log.warn(
205      { err: (err as Error).message, count: notificationIds.length },
206      "emailSentAt stamp failed (email already sent — non-fatal)"
207    );
208  }
209}
210 
211export function createEmailWorker(opts: {
212  connection: Redis;
213  prisma: PrismaClient;
214  env: WorkerEnv;
215  concurrency?: number;
216  smtp?: (o: Parameters<typeof smtpSend>[0]) => Promise<void>;
217}): Worker {
218  const { connection, env } = opts;
219  const send = opts.smtp ?? smtpSend;
220 
221  return new Worker(
222    QUEUES.email,
223    withJobOtel(QUEUES.email, async (job: Job) => {
224      const started = Date.now();
225      const data = emailJobSchema.parse(job.data);
226 
227      if (!env.SMTP_HOST) {
228        log.info(
229          { jobId: job.id, to: data.to, subject: data.subject },
230          "[email] (no SMTP configured, skipping)"
231        );
232        return { skipped: true, reason: "no smtp host" };
233      }
234 
235      const { subject, html, text } = renderEmail(data);
236      await send({
237        host: env.SMTP_HOST,
238        port: env.SMTP_PORT,
239        user: env.SMTP_USER,
240        pass: env.SMTP_PASS,
241        from: env.SMTP_FROM,
242        to: [data.to, ...(data.cc ?? [])],
243        subject,
244        html,
245        text,
246        startTls: env.SMTP_STARTTLS
247      });
248      // TASK-0094 R6: stamp the source Notification row(s) `emailSentAt` on a
249      // REAL send success only (the no-SMTP path above never reaches here). This
250      // is the idempotency flag — a job redelivered after a partial SMTP
251      // failure re-runs the send, but the decision consumer + this stamp stop
252      // the double-send. Stamping must not fail the job (the mail already went
253      // out), so it is best-effort.
254      if (data.notificationIds?.length) {
255        await stampEmailSent(opts.prisma, data.notificationIds);
256      }
257      log.info(
258        { jobId: job.id, to: data.to, subject, ms: Date.now() - started },
259        "email sent"
260      );
261    }),
262    {
263      connection,
264      concurrency: opts.concurrency ?? env.WORKER_CONCURRENCY
265    }
266  );
267}
268