1import { describe, expect, it, vi } from "vitest";
2import type { PrismaClient } from "@prisma/client";
3import {
4 buildInstantEmailData,
5 collectDigest,
6 decideEmail,
7 handleNotificationCreated,
8 isSelfNotification,
9 notificationCreatedPayloadSchema,
10 resolveDeepLink,
11 runDigestRun,
12 DIGEST_MAX_ITEMS,
13 type NotificationCreatedPayload
14} from "../processors/email-decision.processor.js";
15import { renderEmail, emailJobSchema, stampEmailSent } from "../processors/email.processor.js";
16import { parseEvent, tryParseEvent } from "../events.js";
17import type { JobsOptions } from "bullmq";
18
19const ENV = { PUBLIC_URL: "http://factory.test" };
20const WS = { slug: "acme", id: "ws1", name: "Acme" };
21
22
23
24
25interface FakePrismaOpts {
26 notification?: (where: { id: string }) => unknown;
27 user?: (where: { id: string }) => unknown;
28 workspace?: (where: { id: string }) => unknown;
29 databaseRecord?: (where: { id: string }) => unknown;
30 comment?: (where: { id: string }) => unknown;
31 notificationMany?: () => unknown[];
32}
33
34function fakePrisma(opts: FakePrismaOpts = {}) {
35 const updateMany = vi.fn(async () => ({ count: 1 }));
36 const prisma = {
37 notification: {
38 findUnique: vi.fn(async (a: { where: { id: string } }) => opts.notification?.(a.where) ?? null),
39 findMany: vi.fn(async () => opts.notificationMany?.() ?? []),
40 updateMany
41 },
42 user: { findUnique: vi.fn(async (a: { where: { id: string } }) => opts.user?.(a.where) ?? null) },
43 workspace: { findUnique: vi.fn(async (a: { where: { id: string } }) => opts.workspace?.(a.where) ?? null) },
44 databaseRecord: {
45 findUnique: vi.fn(async (a: { where: { id: string } }) => opts.databaseRecord?.(a.where) ?? null)
46 },
47 comment: { findUnique: vi.fn(async (a: { where: { id: string } }) => opts.comment?.(a.where) ?? null) }
48 };
49 return { prisma: prisma as unknown as PrismaClient, updateMany };
50}
51
52type EnqCall = { queue: string; data: Record<string, unknown>; opts: JobsOptions | undefined };
53function fakeEnqueue() {
54 const calls: EnqCall[] = [];
55
56
57 const fn = async (
58 _c: unknown,
59 queue: string,
60 data: Record<string, unknown>,
61 opts?: JobsOptions
62 ) => {
63 calls.push({ queue, data, opts });
64 return {} as never;
65 };
66 return { fn, calls };
67}
68
69
70
71
72describe("decideEmail", () => {
73 const user = { id: "u1" };
74
75 it("MENTION with default prefs → instant", () => {
76 expect(decideEmail({ type: "MENTION", emailSentAt: null, actorId: "u2" }, user, {})).toEqual({
77 kind: "instant"
78 });
79 });
80
81 it("COMMENT with default prefs → digest", () => {
82 expect(decideEmail({ type: "COMMENT", emailSentAt: null, actorId: "u2" }, user, {})).toEqual({
83 kind: "digest"
84 });
85 });
86
87 it("pref 'off' → skip(pref-off)", () => {
88 expect(
89 decideEmail({ type: "MENTION", emailSentAt: null, actorId: "u2" }, user, { MENTION: "off" })
90 ).toEqual({ kind: "skip", reason: "pref-off" });
91 });
92
93 it("pref 'all' → instant", () => {
94 expect(
95 decideEmail({ type: "COMMENT", emailSentAt: null, actorId: "u2" }, user, { COMMENT: "all" })
96 ).toEqual({ kind: "instant" });
97 });
98
99 it("already emailed (emailSentAt set) → skip(already-sent)", () => {
100 expect(
101 decideEmail({ type: "MENTION", emailSentAt: new Date(), actorId: "u2" }, user, {})
102 ).toEqual({ kind: "skip", reason: "already-sent" });
103 });
104
105 it("user gone → skip(user-not-found)", () => {
106 expect(
107 decideEmail({ type: "MENTION", emailSentAt: null, actorId: "u2" }, null, {})
108 ).toEqual({ kind: "skip", reason: "user-not-found" });
109 });
110
111 it("self-notification (actorId === userId) → skip(self-notification)", () => {
112 expect(
113 decideEmail({ type: "MENTION", emailSentAt: null, actorId: "u1" }, { id: "u1" }, {})
114 ).toEqual({ kind: "skip", reason: "self-notification" });
115 });
116
117 it("unknown type → skip(unknown-type)", () => {
118 expect(decideEmail({ type: "WEIRD", emailSentAt: null, actorId: "u2" }, user, {})).toEqual({
119 kind: "skip",
120 reason: "unknown-type"
121 });
122 });
123
124 it("malformed stored prefs fall back to defaults (never throw)", () => {
125 expect(decideEmail({ type: "MENTION", emailSentAt: null, actorId: "u2" }, user, "garbage")).toEqual(
126 { kind: "instant" }
127 );
128 });
129
130 it("partial stored prefs merge over defaults (COMMENT override → instant)", () => {
131 expect(decideEmail({ type: "COMMENT", emailSentAt: null, actorId: "u2" }, user, { COMMENT: "instant" })).toEqual(
132 { kind: "instant" }
133 );
134 });
135});
136
137describe("isSelfNotification", () => {
138 it("true when actorId === userId", () => {
139 expect(isSelfNotification("u1", "u1")).toBe(true);
140 });
141 it("false on a different actor or null", () => {
142 expect(isSelfNotification("u2", "u1")).toBe(false);
143 expect(isSelfNotification(null, "u1")).toBe(false);
144 });
145});
146
147
148
149
150describe("buildInstantEmailData", () => {
151 it("builds title + actor + absolute link, carries the notificationIds", () => {
152 const d = buildInstantEmailData({
153 to: "u@x.io",
154 workspaceName: "Acme",
155 title: "Bob mentioned you",
156 actorName: "Bob",
157 link: { type: "page", id: "p1" },
158 url: "http://factory.test/w/acme/page/p1",
159 notificationId: "n1"
160 });
161 expect(d.to).toBe("u@x.io");
162 expect(d.subject).toBe("Acme: Bob mentioned you");
163 expect(d.notificationIds).toEqual(["n1"]);
164 expect(String(d.html)).toContain("http://factory.test/w/acme/page/p1");
165 expect(String(d.html)).toContain("Bob");
166 expect(String(d.text)).toContain("Bob mentioned you");
167 });
168
169 it("omits the link line when the target is unresolvable", () => {
170 const d = buildInstantEmailData({
171 to: "u@x.io",
172 workspaceName: "Acme",
173 title: "T",
174 actorName: null,
175 link: null,
176 url: null,
177 notificationId: "n2"
178 });
179 expect(String(d.html)).toContain("(open the workspace for details)");
180 });
181});
182
183
184
185
186describe("resolveDeepLink", () => {
187 const ws = () => fakePrisma({ workspace: () => WS });
188
189 it("page → /w/{slug}/page/{id}", async () => {
190 expect(await resolveDeepLink(ws().prisma, ENV, "ws1", { type: "page", id: "p1" })).toBe(
191 "http://factory.test/w/acme/page/p1"
192 );
193 });
194
195 it("database → /w/{slug}/databases/{id}", async () => {
196 expect(await resolveDeepLink(ws().prisma, ENV, "ws1", { type: "database", id: "db1" })).toBe(
197 "http://factory.test/w/acme/databases/db1"
198 );
199 });
200
201 it("record → its database", async () => {
202 const { prisma } = fakePrisma({
203 workspace: () => WS,
204 databaseRecord: () => ({ databaseId: "db1" })
205 });
206 expect(await resolveDeepLink(prisma, ENV, "ws1", { type: "record", id: "r1" })).toBe(
207 "http://factory.test/w/acme/databases/db1"
208 );
209 });
210
211 it("comment → its page", async () => {
212 const { prisma } = fakePrisma({
213 workspace: () => WS,
214 comment: () => ({ pageId: "p1", recordId: null })
215 });
216 expect(await resolveDeepLink(prisma, ENV, "ws1", { type: "comment", id: "c1" })).toBe(
217 "http://factory.test/w/acme/page/p1"
218 );
219 });
220
221 it("comment → its record's database (no page)", async () => {
222 const { prisma } = fakePrisma({
223 workspace: () => WS,
224 comment: () => ({ pageId: null, recordId: "r1" }),
225 databaseRecord: () => ({ databaseId: "db9" })
226 });
227 expect(await resolveDeepLink(prisma, ENV, "ws1", { type: "comment", id: "c1" })).toBe(
228 "http://factory.test/w/acme/databases/db9"
229 );
230 });
231
232 it("missing workspace → null", async () => {
233 const { prisma } = fakePrisma({});
234 expect(await resolveDeepLink(prisma, ENV, "ghost", { type: "page", id: "p1" })).toBeNull();
235 });
236
237 it("no link → workspace root; unknown type → workspace root", async () => {
238 expect(await resolveDeepLink(ws().prisma, ENV, "ws1", null)).toBe("http://factory.test/w/acme");
239 expect(await resolveDeepLink(ws().prisma, ENV, "ws1", { type: "mystery", id: "x" })).toBe(
240 "http://factory.test/w/acme"
241 );
242 });
243
244 it("strips trailing slash from PUBLIC_URL", async () => {
245 expect(await resolveDeepLink(ws().prisma, { PUBLIC_URL: "http://factory.test/" }, "ws1", { type: "page", id: "p1" })).toBe(
246 "http://factory.test/w/acme/page/p1"
247 );
248 });
249});
250
251
252
253
254describe("collectDigest", () => {
255 const row = (i: number, type = "COMMENT") => ({
256 id: `n${i}`,
257 type,
258 title: `T${i}`,
259 link: { type: "page", id: `p${i}` },
260 actorId: "u2"
261 });
262
263 it("caps at 20 and reports the remainder", async () => {
264 const { prisma } = fakePrisma({ workspace: () => WS });
265
266 const rows = Array.from({ length: 25 }, (_, i) => row(i));
267 const { items, remaining } = await collectDigest(prisma, ENV, {
268 userId: "u1",
269 workspaceId: "ws1",
270 storedPrefs: {},
271 rows
272 });
273 expect(items.length).toBe(DIGEST_MAX_ITEMS);
274 expect(remaining).toBe(5);
275 expect(items[0]).toMatchObject({ notificationId: "n0", title: "T0" });
276 expect(items[0]?.url).toBe("http://factory.test/w/acme/page/p0");
277 });
278
279 it("excludes non-digest types ('all' / 'off' overrides)", async () => {
280 const { prisma } = fakePrisma({ workspace: () => WS });
281 const rows = [row(0, "COMMENT"), row(1, "COMMENT"), row(2, "COMMENT")];
282 const { items } = await collectDigest(prisma, ENV, {
283 userId: "u1",
284 workspaceId: "ws1",
285 storedPrefs: {},
286 rows
287 });
288 expect(items.map((i) => i.notificationId)).toEqual(["n0", "n1", "n2"]);
289
290
291 const all = await collectDigest(prisma, ENV, {
292 userId: "u1",
293 workspaceId: "ws1",
294 storedPrefs: { COMMENT: "all" },
295 rows: [row(0, "COMMENT")]
296 });
297 expect(all.items).toHaveLength(0);
298 });
299});
300
301
302
303
304describe("runDigestRun", () => {
305 it("enqueues ONE digest email per (user, workspace) with the collected ids", async () => {
306 const rows = [
307 { id: "n1", userId: "u1", workspaceId: "ws1", type: "COMMENT", title: "A comment", link: { type: "page", id: "p1" }, actorId: "u2" },
308 { id: "n2", userId: "u1", workspaceId: "ws1", type: "DATABASE_CHANGE", title: "A db change", link: { type: "database", id: "db1" }, actorId: "u2" }
309 ];
310 const { prisma } = fakePrisma({
311 notificationMany: () => rows,
312 user: () => ({ email: "u@x.io", notificationPrefs: {} }),
313 workspace: () => WS
314 });
315 const { fn, calls } = fakeEnqueue();
316 const res = await runDigestRun(prisma, "conn" as never, ENV, fn as never);
317 expect(res.enqueued).toBe(1);
318 expect(calls).toHaveLength(1);
319 expect(calls[0]?.queue).toBe("email");
320 expect(calls[0]?.opts?.jobId).toBe("digest:u1:ws1");
321 const data = calls[0]?.data as {
322 to: string;
323 subject: string;
324 template: string;
325 vars: { items: string[]; url: string };
326 notificationIds: string[];
327 };
328 expect(data.to).toBe("u@x.io");
329 expect(data.subject).toBe("Your Factory digest");
330 expect(data.template).toBe("digest");
331 expect(data.notificationIds).toEqual(["n1", "n2"]);
332 expect(data.vars.url).toBe("http://factory.test/w/acme/notifications");
333 expect(data.vars.items).toContain("u2 — A comment");
334 });
335
336 it("does nothing when there are no unsent rows (idempotent re-run)", async () => {
337 const { prisma } = fakePrisma({ notificationMany: () => [] });
338 const { fn, calls } = fakeEnqueue();
339 const res = await runDigestRun(prisma, "conn" as never, ENV, fn as never);
340 expect(res.enqueued).toBe(0);
341 expect(calls).toHaveLength(0);
342 });
343
344 it("appends a '+ n more' line past the 20-item cap", async () => {
345 const rows = Array.from({ length: 25 }, (_, i) => ({
346 id: `n${i}`,
347 userId: "u1",
348 workspaceId: "ws1",
349 type: "COMMENT",
350 title: `T${i}`,
351 link: { type: "page", id: `p${i}` },
352 actorId: "u2"
353 }));
354 const { prisma } = fakePrisma({
355 notificationMany: () => rows,
356 user: () => ({ email: "u@x.io", notificationPrefs: {} }),
357 workspace: () => WS
358 });
359 const { fn, calls } = fakeEnqueue();
360 await runDigestRun(prisma, "conn" as never, ENV, fn as never);
361 const data = calls[0]?.data as { notificationIds: string[]; vars: { items: string[] } };
362 expect(data.notificationIds).toHaveLength(DIGEST_MAX_ITEMS);
363 expect(data.vars.items.at(-1)).toBe("+ 5 more (open your inbox)");
364 });
365});
366
367
368
369
370const payload = (over: Partial<NotificationCreatedPayload> = {}): NotificationCreatedPayload =>
371 ({
372 workspaceId: "ws1",
373 notificationId: "n1",
374 userId: "u1",
375 type: "MENTION",
376 title: "Bob mentioned you",
377 deepLink: { type: "page", id: "p1" },
378 actorName: "Bob",
379 ...over
380 }) as NotificationCreatedPayload;
381
382describe("handleNotificationCreated", () => {
383 it("MENTION + default prefs → enqueues an instant email with the job key", async () => {
384 const { prisma } = fakePrisma({
385 notification: () => ({ type: "MENTION", emailSentAt: null, actorId: "u2", userId: "u1" }),
386 user: () => ({ id: "u1", email: "u@x.io", notificationPrefs: {} }),
387 workspace: () => WS
388 });
389 const { fn, calls } = fakeEnqueue();
390 const res = await handleNotificationCreated(prisma, "conn" as never, ENV, payload(), fn);
391 expect(res.action).toBe("instant");
392 expect(calls).toHaveLength(1);
393 expect(calls[0]?.queue).toBe("email");
394
395
396 expect(calls[0]?.opts?.jobId).toBe("email_n1");
397 expect(calls[0]?.data.notificationIds).toEqual(["n1"]);
398 expect(calls[0]?.data.subject).toBe("Acme: Bob mentioned you");
399 expect(String(calls[0]?.data.html)).toContain("http://factory.test/w/acme/page/p1");
400 });
401
402 it("COMMENT + digest → does NOT enqueue (left for the hourly digest)", async () => {
403 const { prisma } = fakePrisma({
404 notification: () => ({ type: "COMMENT", emailSentAt: null, actorId: "u2", userId: "u1" }),
405 user: () => ({ id: "u1", email: "u@x.io", notificationPrefs: {} })
406 });
407 const { fn, calls } = fakeEnqueue();
408 const res = await handleNotificationCreated(prisma, "conn" as never, ENV, payload({ type: "COMMENT" }), fn);
409 expect(res.action).toBe("digest");
410 expect(calls).toHaveLength(0);
411 });
412
413 it("self-notification → no enqueue (suppression)", async () => {
414 const { prisma } = fakePrisma({
415 notification: () => ({ type: "MENTION", emailSentAt: null, actorId: "u1", userId: "u1" }),
416 user: () => ({ id: "u1", email: "u@x.io", notificationPrefs: {} })
417 });
418 const { fn, calls } = fakeEnqueue();
419 const res = await handleNotificationCreated(prisma, "conn" as never, ENV, payload(), fn);
420 expect(res).toEqual({ action: "skip", reason: "self-notification" });
421 expect(calls).toHaveLength(0);
422 });
423
424 it("already emailed → no enqueue (the redelivery guard)", async () => {
425 const { prisma } = fakePrisma({
426 notification: () => ({ type: "MENTION", emailSentAt: new Date(), actorId: "u2", userId: "u1" }),
427 user: () => ({ id: "u1", email: "u@x.io", notificationPrefs: {} })
428 });
429 const { fn, calls } = fakeEnqueue();
430 const res = await handleNotificationCreated(prisma, "conn" as never, ENV, payload(), fn);
431 expect(res).toEqual({ action: "skip", reason: "already-sent" });
432 expect(calls).toHaveLength(0);
433 });
434
435 it("row gone → skip(row-not-found), no enqueue", async () => {
436 const { prisma } = fakePrisma({});
437 const { fn, calls } = fakeEnqueue();
438 const res = await handleNotificationCreated(prisma, "conn" as never, ENV, payload(), fn);
439 expect(res).toEqual({ action: "skip", reason: "row-not-found" });
440 expect(calls).toHaveLength(0);
441 });
442});
443
444
445
446
447describe("notificationCreatedPayloadSchema", () => {
448 it("accepts a valid payload and passes through extra fields", () => {
449 const ok = notificationCreatedPayloadSchema.safeParse(payload());
450 expect(ok.success).toBe(true);
451 if (ok.success) expect(ok.data.deepLink).toEqual({ type: "page", id: "p1" });
452 });
453 it("rejects a payload missing required fields", () => {
454 const bad = notificationCreatedPayloadSchema.safeParse({ workspaceId: "ws1" });
455 expect(bad.success).toBe(false);
456 });
457});
458
459
460
461
462describe("email.processor", () => {
463 it("digest template: subject 'Your Factory digest', lists items, inbox link", () => {
464 const r = renderEmail({
465 to: "u@x.io",
466 subject: "Your Factory digest",
467 template: "digest",
468 vars: {
469 items: ["Bob — hi", "+ 3 more (open your inbox)"],
470 url: "http://factory.test/w/acme/notifications"
471 }
472 });
473 expect(r.subject).toBe("Your Factory digest");
474 expect(r.html).toContain("Your Factory digest");
475 expect(r.html).toContain("Bob — hi");
476 expect(r.text).toContain("- Bob — hi");
477 expect(r.html).toContain("http://factory.test/w/acme/notifications");
478 });
479
480 it("emailJobSchema is backward-compatible (old shape without notificationIds)", () => {
481 expect(emailJobSchema.safeParse({ to: "u@x.io", subject: "hi", text: "t" }).success).toBe(true);
482 expect(emailJobSchema.safeParse({ to: "u@x.io", subject: "hi", notificationIds: ["n1", "n2"] }).success).toBe(true);
483 });
484
485 it("stampEmailSent marks rows (only unsent) and swallows errors", async () => {
486 const { prisma, updateMany } = fakePrisma({});
487 await stampEmailSent(prisma, ["n1", "n2"]);
488 expect(updateMany).toHaveBeenCalledWith({
489 where: { id: { in: ["n1", "n2"] }, emailSentAt: null },
490 data: { emailSentAt: expect.any(Date) }
491 });
492 updateMany.mockRejectedValueOnce(new Error("db down"));
493 await expect(stampEmailSent(prisma, ["n1"])).resolves.toBeUndefined();
494 });
495});
496
497
498
499
500describe("parseEvent (regression)", () => {
501 it("parses a valid envelope (was: ReferenceError → every entry dropped)", () => {
502 const r = parseEvent({
503 id: "1-0",
504 fields: {
505 event: "notification.created",
506 payload: JSON.stringify({ notificationId: "n1" }),
507 ts: new Date().toISOString()
508 }
509 });
510 expect(r.envelope.event).toBe("notification.created");
511 expect(r.envelope.payload).toEqual({ notificationId: "n1" });
512 });
513 it("drops a malformed envelope (returns null)", () => {
514 expect(
515 tryParseEvent({ id: "1-1", fields: { event: "", payload: "not-json", ts: "t" } })
516 ).toBeNull();
517 });
518});
519
