email-decision.processor.ts

email-decision.processor.tsapps/worker/src/processors/email-decision.processor.tsraw ← processors19,707 B← Back to the app
1/**
2 * apps/worker — notification → email decision (TASK-0094 R6).
3 *
4 * The API creates a `Notification` row (channel IN_APP) and publishes
5 * `notification.created` to the domain-event stream. THIS consumer owns the
6 * whole email channel — the API holds no BullMQ (architecture: the worker owns
7 * every queue). On `notification.created` it decides, per user preference:
8 *
9 *   skip            — unknown type / already emailed (`emailSentAt`) / user
10 *                     gone / self-notification (`actorId === userId`) /
11 *                     pref "off"
12 *   ENQUEUE (instant) — pref "instant" or "all": one email job on the `email`
13 *                     queue, jobId `email_{notificationId}` (the BullMQ
14 *                     job-key dedup — a double-enqueue produces ONE job, so a
15 *                     double-send is impossible at the queue layer). NO colon:
16 *                     BullMQ 5.x rejects custom jobIds containing `:` unless
17 *                     exactly 3-segment (repeatable compat) — the old
18 *                     `email:{id}` threw on every enqueue.
19 *   DIGEST          — pref "digest": left for the hourly digest job
20 *
21 * The HOURLY DIGEST (a repeatable `maintenance` job, `notification-digest`)
22 * collects each user's unsent, digest-eligible notifications (≤ 20 per email,
23 * "n more" inbox link for the rest), sends one email per (user, workspace),
24 * and stamps the collected rows `emailSentAt` so the next run never re-sends
25 * them. At-least-once delivery is why the double-send guard is the
26 * `emailSentAt` column + the job key, NOT the job key alone: a job redelivered
27 * after a partial SMTP failure re-runs the processor, and only the durable
28 * stamp (checked here at decision time AND by the email worker before sending)
29 * closes that window. (Recorded conflict: the spec's Database-Changes note
30 * assumed the job key was sufficient and that a Phase-1 delivery log existed —
31 * it does not; see the task file.)
32 *
33 * Deep links are ABSOLUTE (`PUBLIC_URL` env — the spec's name; the API calls
34 * the same value `WEB_ORIGIN`, recorded) and require login on the web side
35 * (a link to a deleted target → the standard 404, not a data leak).
36 */
37import type { Redis } from "ioredis";
38import { z } from "zod";
39import type { PrismaClient } from "@prisma/client";
40import {
41  NOTIFICATION_TYPES,
42  mergeNotificationPrefs,
43  prefIsOff,
44  prefSendsInstant,
45  type NotificationTypeValue
46} from "@factory/shared";
47import { QUEUES, enqueue } from "../queues.js";
48import { StreamConsumer, type StreamEntry } from "../lib/redis-stream.js";
49import { tryParseEvent } from "../events.js";
50import { getLogger } from "../logger.js";
51 
52const log = getLogger("email-decision");
53 
54// ---------------------------------------------------------------------------
55// Payload + decision (pure — the unit-testable core)
56// ---------------------------------------------------------------------------
57 
58/**
59 * The notification's deep link. The API's `notification.created` payload
60 * carries it as `deepLink`; the row itself stores the same object as `link` —
61 * accept either (`.passthrough()` on the payload keeps the rest of the map).
62 */
63const notificationLinkSchema = z.object({
64  deepLink: z.object({ type: z.string().min(1), id: z.string().min(1) }),
65  link: z
66    .object({ type: z.string().min(1), id: z.string().min(1) })
67    .optional()
68});
69export type NotificationLink = { type: string; id: string };
70 
71/** The `notification.created` stream payload (API `notifications.service`). */
72export const notificationCreatedPayloadSchema = z
73  .object({
74    workspaceId: z.string().min(1),
75    notificationId: z.string().min(1),
76    userId: z.string().min(1),
77    type: z.string().min(1),
78    title: z.string().min(1)
79  })
80  .passthrough();
81export type NotificationCreatedPayload = z.infer<typeof notificationCreatedPayloadSchema>;
82 
83/** The `actorId == notification.userId → no email` suppression (R6). */
84export function isSelfNotification(actorId: string | null, userId: string): boolean {
85  return actorId !== null && actorId === userId;
86}
87 
88export type EmailDecision =
89  | { kind: "instant" }
90  | { kind: "digest" }
91  | { kind: "skip"; reason: string };
92 
93/**
94 * The per-notification email decision (pure). `prefs` is the user's STORED
95 * (partial, unvalidated) prefs — merged over the shared defaults here, the
96 * same helper the API `GET /notifications/preferences` uses, so the two can
97 * never diverge.
98 */
99export function decideEmail(
100  notification: {
101    type: string;
102    emailSentAt: Date | null;
103    actorId: string | null;
104  },
105  user: { id: string } | null,
106  prefs: unknown
107): EmailDecision {
108  const type = notification.type;
109  if (!(NOTIFICATION_TYPES as readonly string[]).includes(type)) {
110    return { kind: "skip", reason: "unknown-type" };
111  }
112  const t = type as NotificationTypeValue;
113  if (notification.emailSentAt) return { kind: "skip", reason: "already-sent" };
114  if (!user) return { kind: "skip", reason: "user-not-found" };
115  if (isSelfNotification(notification.actorId, user.id)) {
116    return { kind: "skip", reason: "self-notification" };
117  }
118  const merged = mergeNotificationPrefs(prefs);
119  if (prefIsOff(t, merged)) return { kind: "skip", reason: "pref-off" };
120  if (prefSendsInstant(t, merged)) return { kind: "instant" };
121  return { kind: "digest" };
122}
123 
124/**
125 * Build the instant email job's `data` (title + actor + absolute link only —
126 * never the comment body: the deep link is the path to it, a smaller leak
127 * surface, recorded). `notificationId` is carried so the email worker stamps
128 * `Notification.emailSentAt` on a successful send.
129 */
130export function buildInstantEmailData(input: {
131  to: string;
132  workspaceName: string;
133  title: string;
134  actorName: string | null;
135  link: NotificationLink | null;
136  url: string | null;
137  notificationId: string;
138}): Record<string, unknown> {
139  const { to, workspaceName, title, actorName, url, notificationId } = input;
140  const from = actorName ? `From <strong>${esc(actorName)}</strong>. ` : "";
141  const line = url
142    ? `<p><a href="${esc(url)}">Open in workspace</a></p>`
143    : "";
144  const html =
145    `<p>${from}${esc(title)}</p>` +
146    (url ? line : `<p>(open the workspace for details)</p>`) +
147    `<p style="color:#999;font-size:12px">${esc(workspaceName)}</p>`;
148  const text =
149    `${actorName ? `${actorName} — ` : ""}${title}\n` +
150    (url ? url : "") +
151    `\n${workspaceName}`;
152  return {
153    to,
154    subject: `${workspaceName}: ${title}`,
155    html,
156    text,
157    notificationIds: [notificationId]
158  };
159}
160 
161function esc(s: string): string {
162  return s.replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;");
163}
164 
165// ---------------------------------------------------------------------------
166// Absolute deep links
167// ---------------------------------------------------------------------------
168 
169/**
170 * Resolve a notification deep link to an ABSOLUTE web URL (the spec's
171 * `PUBLIC_URL` env). Web routes: `/w/{slugOrId}/page/{pageId}`,
172 * `/w/{slugOrId}/databases/{databaseId}`, inbox `/w/{slugOrId}/notifications`.
173 *
174 *   page     → /w/{ws}/page/{id}
175 *   database → /w/{ws}/databases/{id}
176 *   record   → its database (the record's home) → /w/{ws}/databases/{dbId}
177 *   comment  → its page, else its record's database
178 *
179 * Unresolvable (deleted target, unknown type, missing workspace) → the
180 * workspace root (still requires login; a deleted target → the web's 404).
181 * Never throws — an email must never fail because of a stale link.
182 */
183export async function resolveDeepLink(
184  prisma: PrismaClient,
185  env: { PUBLIC_URL: string },
186  workspaceId: string,
187  link: NotificationLink | null
188): Promise<string | null> {
189  const ws = await prisma.workspace.findUnique({
190    where: { id: workspaceId },
191    select: { slug: true, id: true }
192  });
193  if (!ws) return null;
194  const wsPart = ws.slug || ws.id;
195  const base = env.PUBLIC_URL.replace(/\/+$/, "");
196  if (!link) return `${base}/w/${wsPart}`;
197  switch (link.type) {
198    case "page":
199      return `${base}/w/${wsPart}/page/${link.id}`;
200    case "database":
201      return `${base}/w/${wsPart}/databases/${link.id}`;
202    case "record": {
203      const rec = await prisma.databaseRecord.findUnique({
204        where: { id: link.id },
205        select: { databaseId: true }
206      });
207      return rec
208        ? `${base}/w/${wsPart}/databases/${rec.databaseId}`
209        : `${base}/w/${wsPart}`;
210    }
211    case "comment": {
212      const c = await prisma.comment.findUnique({
213        where: { id: link.id },
214        select: { pageId: true, recordId: true }
215      });
216      if (c?.pageId) return `${base}/w/${wsPart}/page/${c.pageId}`;
217      if (c?.recordId) {
218        const rec = await prisma.databaseRecord.findUnique({
219          where: { id: c.recordId },
220          select: { databaseId: true }
221        });
222        return rec ? `${base}/w/${wsPart}/databases/${rec.databaseId}` : `${base}/w/${wsPart}`;
223      }
224      return `${base}/w/${wsPart}`;
225    }
226    default:
227      return `${base}/w/${wsPart}`;
228  }
229}
230 
231/** Absolute inbox URL (the digest "n more" link). */
232export async function inboxUrl(
233  prisma: PrismaClient,
234  env: { PUBLIC_URL: string },
235  workspaceId: string
236): Promise<string | null> {
237  const ws = await prisma.workspace.findUnique({
238    where: { id: workspaceId },
239    select: { slug: true, id: true }
240  });
241  if (!ws) return null;
242  return `${env.PUBLIC_URL.replace(/\/+$/, "")}/w/${ws.slug || ws.id}/notifications`;
243}
244 
245// ---------------------------------------------------------------------------
246// Digest collection (pure — the ≤ 20 cap + "n more" link)
247// ---------------------------------------------------------------------------
248 
249export interface DigestItem {
250  notificationId: string;
251  title: string;
252  actorName: string | null;
253  url: string | null;
254  link: NotificationLink | null;
255}
256 
257/**
258 * Turn a user's (unsent) digest-eligible notifications in ONE workspace into
259 * digest email items. Only types whose effective pref mode is exactly "digest"
260 * qualify — an "all"/"instant" type already received an INSTANT email
261 * (→ `emailSentAt` stamped → excluded by the caller's query anyway), so
262 * including it here would risk a double-send in the pre-stamp race window;
263 * "off"/unknown are excluded. Each link is resolved to an absolute URL
264 * (best-effort). Capped at {@link DIGEST_MAX_ITEMS} per email. Pure except for
265 * the link-resolution reads — unit-tested with a fake prisma.
266 *
267 * (Recorded: the spec's `'all'` mode is interpreted as "send immediately" —
268 * matching `prefSendsInstant` which returns true for both "instant" and "all".
269 * "all" therefore never batches into the digest; the instant path covers it.)
270 */
271export const DIGEST_MAX_ITEMS = 20;
272 
273export async function collectDigest(
274  prisma: PrismaClient,
275  env: { PUBLIC_URL: string },
276  input: {
277    userId: string;
278    workspaceId: string;
279    storedPrefs: unknown;
280    rows: Array<{
281      id: string;
282      type: string;
283      title: string;
284      link: unknown;
285      actorId: string | null;
286    }>;
287  }
288): Promise<{ items: DigestItem[]; remaining: number }> {
289  const merged = mergeNotificationPrefs(input.storedPrefs);
290  const eligible: typeof input.rows = [];
291  for (const row of input.rows) {
292    const t = row.type as NotificationTypeValue;
293    if (!(NOTIFICATION_TYPES as readonly string[]).includes(row.type)) continue;
294    if (merged[t] !== "digest") continue;
295    eligible.push(row);
296  }
297  const capped = eligible.slice(0, DIGEST_MAX_ITEMS);
298  const remaining = eligible.length - capped.length;
299  const items: DigestItem[] = [];
300  for (const row of capped) {
301    const link = (row.link ?? null) as NotificationLink | null;
302    const url = link ? await resolveDeepLink(prisma, env, input.workspaceId, link) : null;
303    items.push({
304      notificationId: row.id,
305      title: row.title,
306      actorName: row.actorId ?? null,
307      url,
308      link
309    });
310  }
311  return { items, remaining };
312}
313 
314// ---------------------------------------------------------------------------
315// Digest runner (hourly)
316// ---------------------------------------------------------------------------
317 
318/**
319 * The hourly digest: for every (user, workspace) pair with ≥1 unread, unsent,
320 * digest-eligible notification, enqueue ONE email (template "digest", subject
321 * "Your Factory digest", ≤ {@link DIGEST_MAX_ITEMS} items + "n more" inbox
322 * link) carrying the collected `notificationIds` so the email worker stamps
323 * `emailSentAt` on a successful send (never re-sent next hour). Returns the
324 * number of digest emails enqueued.
325 *
326 * `enqueue` is injectable for tests. Idempotent across the redelivery of the
327 * repeatable job itself: a second run finds no unsent rows (already stamped)
328 * → enqueues nothing.
329 */
330export async function runDigestRun(
331  prisma: PrismaClient,
332  connection: Redis,
333  env: { PUBLIC_URL: string },
334  enqueueFn?: typeof enqueue
335): Promise<{ enqueued: number }> {
336  const enq = enqueueFn ?? enqueue;
337  const rows = await prisma.notification.findMany({
338    where: { readAt: null, emailSentAt: null },
339    select: {
340      id: true,
341      userId: true,
342      workspaceId: true,
343      type: true,
344      title: true,
345      link: true,
346      actorId: true
347    }
348  });
349  // Group by (userId, workspaceId).
350  const byUser = new Map<
351    string,
352    { userId: string; workspaceId: string; rows: typeof rows }
353  >();
354  for (const r of rows) {
355    const key = `${r.userId}\u0000${r.workspaceId}`;
356    let g = byUser.get(key);
357    if (!g) {
358      g = { userId: r.userId, workspaceId: r.workspaceId, rows: [] };
359      byUser.set(key, g);
360    }
361    g.rows.push(r);
362  }
363 
364  let enqueued = 0;
365  for (const g of byUser.values()) {
366    const user = await prisma.user.findUnique({
367      where: { id: g.userId },
368      select: { email: true, notificationPrefs: true }
369    });
370    if (!user) continue;
371    const { items, remaining } = await collectDigest(prisma, env, {
372      userId: g.userId,
373      workspaceId: g.workspaceId,
374      storedPrefs: user.notificationPrefs,
375      rows: g.rows
376    });
377    if (items.length === 0) continue;
378 
379    const inbox = await inboxUrl(prisma, env, g.workspaceId);
380    const bodyItems = items.map((it) => (it.actorName ? `${it.actorName} — ${it.title}` : it.title));
381    const vars: Record<string, unknown> = {
382      items: bodyItems,
383      url: inbox ?? ""
384    };
385    if (remaining > 0) {
386      vars.items = [...bodyItems, `+ ${remaining} more (open your inbox)`];
387    }
388    await enq(
389      connection,
390      QUEUES.email,
391      {
392        to: user.email,
393        subject: "Your Factory digest",
394        template: "digest",
395        vars,
396        notificationIds: items.map((it) => it.notificationId)
397      },
398      // Stable job key: one digest email per (user, workspace, run-window) — a
399      // redelivered repeatable tick must not double-enqueue. The window is the
400      // collected set's ids (the stamp makes a repeat a no-op regardless).
401      { jobId: `digest:${g.userId}:${g.workspaceId}` }
402    );
403    enqueued++;
404    log.info(
405      { userId: g.userId, workspaceId: g.workspaceId, items: items.length, remaining },
406      "digest email enqueued"
407    );
408  }
409  return { enqueued };
410}
411 
412// ---------------------------------------------------------------------------
413// Stream consumer (worker-email)
414// ---------------------------------------------------------------------------
415 
416/**
417 * The per-event handler (extracted from the StreamConsumer for unit-testing the
418 * full decision→enqueue path with fakes). `enqueueFn` is injectable; the real
419 * one is the BullMQ `enqueue` helper (dedup by `jobId`). Returns the action
420 * taken (for logs/tests). Never throws on a business skip — a throw here would
421 * leave the stream entry pending for redelivery.
422 */
423export async function handleNotificationCreated(
424  prisma: PrismaClient,
425  connection: Redis,
426  env: { PUBLIC_URL: string },
427  payload: NotificationCreatedPayload,
428  enqueueFn?: typeof enqueue
429): Promise<{ action: "instant" | "digest" | "skip"; reason?: string }> {
430  const enq = enqueueFn ?? enqueue;
431 
432  // Load the authoritative row (the stream payload is a fan-out hint).
433  const row = await prisma.notification.findUnique({
434    where: { id: payload.notificationId },
435    select: { type: true, emailSentAt: true, actorId: true, userId: true }
436  });
437  if (!row) {
438    return { action: "skip", reason: "row-not-found" };
439  }
440  const user = await prisma.user.findUnique({
441    where: { id: row.userId },
442    select: { id: true, email: true, notificationPrefs: true }
443  });
444 
445  const decision = decideEmail(row, user, user?.notificationPrefs ?? null);
446  if (decision.kind === "skip") return { action: "skip", reason: decision.reason };
447  if (decision.kind === "digest") return { action: "digest" }; // hourly digest collects it
448  if (!user) return { action: "skip", reason: "user-not-found" };
449 
450  // instant:
451  const actorName =
452    typeof payload.actorName === "string" && payload.actorName ? payload.actorName : null;
453  const link = (payload.deepLink ?? payload.link) as NotificationLink | undefined;
454  const url = await resolveDeepLink(prisma, env, payload.workspaceId, link ?? null);
455  const ws = await prisma.workspace.findUnique({
456    where: { id: payload.workspaceId },
457    select: { name: true }
458  });
459  const data = buildInstantEmailData({
460    to: user.email,
461    workspaceName: ws?.name ?? "Factory",
462    title: payload.title,
463    actorName,
464    link: link ?? null,
465    url,
466    notificationId: payload.notificationId
467  });
468  await enq(connection, QUEUES.email, data, {
469    // Dedup key = the notification id. NO colon: BullMQ 5.x `Job.validateOptions`
470    // rejects any custom jobId containing `:` (only a 3-segment id is legal —
471    // repeatable compatibility). The old `email:${id}` (2-segment) made EVERY
472    // instant-enqueue throw `Custom Id cannot contain :` (recorded conflict,
473    // FEATURE_TRACKER T-176/177/178 — fixed 2026-10-12). The underscore keeps
474    // the human-readable `email` prefix in the Jobs-tab DLQ view; nothing
475    // parses the jobId prefix (health tab correlates on payload).
476    jobId: `email_${payload.notificationId}`
477  });
478  return { action: "instant" };
479}
480 
481/**
482 * The `worker-email` StreamConsumer (its own consumer group, so email lag can't
483 * block the other queues). Only `notification.created` is actionable. The
484 * consumer only ENqueues — the real send is the `email` queue's job (stamping
485 * `emailSentAt` happens there), so there is no SMTP dependency here.
486 *
487 * `connection` is the BullMQ Redis (for `enqueue`); `redis` drives the stream.
488 */
489export function createNotificationEmailConsumer(opts: {
490  redis: Redis;
491  connection: Redis;
492  prisma: PrismaClient;
493  env: { PUBLIC_URL: string };
494  consumerId: string;
495}): StreamConsumer {
496  const { redis, connection, prisma, env } = opts;
497  return new StreamConsumer(
498    redis,
499    "factory:events",
500    "worker-email",
501    opts.consumerId,
502    async (entry: StreamEntry) => {
503      const parsed = tryParseEvent(entry);
504      if (!parsed) return; // poison entry: already logged; drop + XACK
505      const { envelope } = parsed;
506      if (envelope.event !== "notification.created") return; // XACK: not ours
507      const payload = notificationCreatedPayloadSchema.safeParse(
508        envelope.payload as Record<string, unknown>
509      );
510      if (!payload.success) {
511        log.warn(
512          { id: entry.id, err: payload.error.message },
513          "dropping malformed notification.created payload"
514        );
515        return;
516      }
517      const res = await handleNotificationCreated(prisma, connection, env, payload.data);
518      if (res.action === "instant") {
519        log.info({ notificationId: payload.data.notificationId }, "instant email enqueued");
520      }
521    }
522  );
523}
524