notification-email-jobid.integration.test.ts

notification-email-jobid.integration.test.tstests/integration/notification-email-jobid.integration.test.tsraw ← integration7,139 B← Back to the app
1/**
2 * T-176/177/178 recorded-conflict REGRESSION (fixed 2026-10-12) — the worker's
3 * instant-notification email jobId must be accepted by REAL BullMQ.
4 *
5 * Background: the old dedup key `email:{notificationId}` (two-segment) is
6 * rejected by BullMQ 5.x `Job.validateOptions` (`Custom Id cannot contain :` —
7 * only a 3-segment colon id is legal, for repeatable compatibility). Every
8 * instant email therefore THREW at enqueue: the stream consumer re-delivered
9 * the entry, `emailSentAt` never stamped, and the user never got the email.
10 * The worker's unit test masked the bug — it injects a MOCKED `enqueue` and
11 * only asserts the jobId string, so it stayed green while the real BullMQ
12 * path was broken. This spec is the real-BullMQ half of the contract:
13 *
14 *   1. `handleNotificationCreated` with the REAL `enqueue` (no enqueueFn
15 *      seam) enqueues onto the live test Redis WITHOUT throwing — the exact
16 *      line that used to throw.
17 *   2. A second enqueue with the SAME jobId is deduped by BullMQ (the job
18 *      stays one) — the property the stable key exists for: the at-least-
19 *      once stream redelivers `notification.created`, so the decision runs
20 *      twice, and the job-key must absorb the duplicate.
21 *   3. The `email` worker (injected `smtp` — no mail server needed) processes
22 *      the job and stamps `Notification.emailSentAt` — the pipeline the old
23 *      bug broke end-to-end.
24 */
25import { afterAll, beforeAll, describe, expect, it } from "vitest";
26import { Queue } from "bullmq";
27import Redis from "ioredis";
28import { createTestApp, type TestAppHandle } from "./helpers/api.js";
29import { getRedisUrl } from "./helpers/containers.js";
30import {
31  buildInstantEmailData,
32  handleNotificationCreated
33} from "../../apps/worker/src/processors/email-decision.processor.js";
34import { createEmailWorker } from "../../apps/worker/src/processors/email.processor.js";
35import { QUEUES, enqueue } from "../../apps/worker/src/queues.js";
36import type { WorkerEnv } from "../../apps/worker/src/config.js";
37import type { PrismaClient } from "@prisma/client";
38 
39describe("instant-notification email jobId (real BullMQ)", () => {
40  let app: TestAppHandle;
41  let redis: Redis;
42  let q: Queue;
43  let worker: ReturnType<typeof createEmailWorker>;
44  let prisma: PrismaClient;
45  let notifId: string;
46  let sent: { to: string; subject: string } | null;
47 
48  beforeAll(async () => {
49    app = await createTestApp();
50    prisma = app.prisma;
51    redis = new Redis(await getRedisUrl(), { maxRetriesPerRequest: null });
52    q = new Queue("email", { connection: redis });
53 
54    // SMTP is FAKE: the worker's `smtp` seam stands in for the transport, so
55    // the job runs the REAL send path (schema parse → render → "send" →
56    // emailSentAt stamp) without a mail server. Worker starts PAUSED — the
57    // dedup assertions below need the job in the WAITING state.
58    sent = null;
59    const env = {
60      SMTP_HOST: "smtp.fake",
61      SMTP_PORT: 587,
62      SMTP_FROM: "noreply@factory.test",
63      SMTP_STARTTLS: false
64    } as unknown as WorkerEnv;
65    worker = createEmailWorker({
66      connection: redis,
67      prisma,
68      env,
69      concurrency: 1,
70      smtp: async (o) => {
71        sent = { to: String(o.to[0]), subject: o.subject };
72      }
73    });
74    await worker.pause();
75  }, 180_000);
76 
77  afterAll(async () => {
78    await worker?.close();
79    await q?.close();
80    await redis?.quit().catch(() => {});
81    await app?.close();
82  }, 30_000);
83 
84  it("real enqueue does not throw + dedup on redelivery + worker stamps emailSentAt", async () => {
85    const run = Math.random().toString(36).slice(2, 8);
86    const user = await prisma.user.create({
87      data: { email: `jobid-probe-${run}@example.com`, name: "JobId Probe" }
88    });
89    const actor = await prisma.user.create({
90      data: { email: `jobid-actor-${run}@example.com`, name: "JobId Actor" }
91    });
92    const ws = await prisma.workspace.create({
93      data: { name: `JobId WS ${run}`, slug: `jobid-${run}` }
94    });
95    const notif = await prisma.notification.create({
96      data: {
97        workspaceId: ws.id,
98        userId: user.id,
99        actorId: actor.id,
100        type: "MENTION",
101        title: "JobId probe mention"
102      }
103    });
104    notifId = notif.id;
105 
106    // --- 1. the exact line that used to throw -------------------------------
107    // Real `enqueue` (default, no seam) → real BullMQ `Job.validateOptions`.
108    // Old key `email:${id}`: throws `Custom Id cannot contain :` here.
109    const res = await handleNotificationCreated(
110      prisma,
111      redis,
112      { PUBLIC_URL: "http://factory.test" },
113      {
114        workspaceId: ws.id,
115        notificationId: notif.id,
116        userId: user.id,
117        type: "MENTION",
118        title: "JobId probe mention",
119        deepLink: { type: "page", id: "p-nonexistent" },
120        actorName: "JobId Actor"
121      }
122    );
123    expect(res.action).toBe("instant");
124 
125    const job = await q.getJob(`email_${notif.id}`);
126    expect(job).toBeTruthy();
127 
128    // --- 2. dedup: the stream redelivers notification.created — the second
129    //        decision runs the SAME enqueue and must NOT add a job.
130    const data = buildInstantEmailData({
131      to: user.email,
132      workspaceName: ws.name,
133      title: "JobId probe mention",
134      actorName: "JobId Actor",
135      link: { type: "page", id: "p-nonexistent" },
136      url: "http://factory.test/w/jobid/page/p-nonexistent",
137      notificationId: notif.id
138    });
139    const re = await enqueue(redis, QUEUES.email, data, { jobId: `email_${notif.id}` });
140    expect(re.id).toBe(`email_${notif.id}`);
141    const waiting = await q.getJobs(["waiting"], 0, -1);
142    expect(waiting.filter((j) => j.id === `email_${notif.id}`)).toHaveLength(1);
143 
144    // --- 3. the worker sends + stamps emailSentAt ---------------------------
145    // The worker is PAUSED (needed for the dedup assertions above) — arm the
146    // listener, RESUME, THEN await. (Await-before-resume would deadlock: the
147    // paused worker never completes the waiting job.)
148    const finished: Promise<unknown> = new Promise((resolve, reject) => {
149      const t = setTimeout(
150        () => reject(new Error("email job did not complete in 15 s")),
151        15_000
152      );
153      worker.once("failed", (j, err) => {
154        clearTimeout(t);
155        reject(new Error(`email job FAILED: ${err?.message ?? "?"}`));
156      });
157      worker.once("completed", (j) => {
158        clearTimeout(t);
159        resolve(j);
160      });
161    });
162    await worker.resume();
163    const done = (await finished) as { id: string };
164    expect(done.id).toBe(`email_${notif.id}`);
165    expect(sent?.to).toBe(user.email);
166    expect(sent?.subject).toBeTruthy();
167    const stamped = await prisma.notification.findUniqueOrThrow({
168      where: { id: notif.id },
169      select: { emailSentAt: true }
170    });
171    expect(stamped.emailSentAt).toBeInstanceOf(Date);
172 
173    // cleanup
174    await prisma.notification.deleteMany({ where: { id: notif.id } });
175    await prisma.workspace.deleteMany({ where: { id: ws.id } });
176    await prisma.user.deleteMany({ where: { id: { in: [user.id, actor.id] } } });
177  }, 30_000);
178});
179