notifications.service.ts

notifications.service.tsapps/api/src/modules/org/notifications.service.tsraw ← org6,509 B← Back to the app
1/**
2 * T-178 — org-admin NOTIFICATIONS tab (delivery health).
3 *
4 * The email pipeline is: API creates the Notification row (channel IN_APP) +
5 * publishes `notification.created` → the worker's email-decision consumer
6 * decides + enqueues ONE email job per notification on the `email` queue (its
7 * payload carries `notificationIds: [id]`) → the email processor sends and
8 * stamps `emailSentAt`.
9 *
10 * This service reports, for the org's workspaces:
11 *   - `inAppTotal` / `inAppUnread` — the bell volume + the unread backlog;
12 *   - `emailQueued`  — email-channel jobs in the queue (waiting/delayed/
13 *     active);
14 *   - `emailFailed`  — email-channel jobs in the FAILED state (each carries
15 *     its failedReason + attemptsMade);
16 *   - `failedJobs`   — the recent failed email jobs (the "recent failures"
17 *     list the tab shows), newest first.
18 *
19 * Org-scoping (no-leak) is done in Postgres: a queue job is only surfaced
20 * when its PAYLOAD's `notificationIds` contain a notification in THIS org
21 * (another org's failed emails are never listed). Correlation is by payload,
22 * NOT by the BullMQ job id — the worker's instant-email dedup key is
23 * `email_{id}` (colon-FREE: BullMQ 5.x rejects custom jobIds containing `:`
24 * unless exactly 3-segment; the old `email:{id}` made every instant-enqueue
25 * throw — see the T-178 conflict note, FIXED 2026-10-12). Parseable but
26 * payload correlation stays the robust path (digest jobs carry many ids;
27 * automation/webhook jobs carry none).
28 */
29import { Inject, Injectable } from "@nestjs/common";
30import type { Request } from "express";
31import { PrismaService } from "../../common/prisma.service.js";
32import { RedisService } from "../../common/redis.service.js";
33import { requireUser } from "../../common/types.js";
34import { assertOrgAdmin } from "./guards.js";
35 
36/** Cap on the recent-failures list (one screenful). */
37const FAILED_LIST_CAP = 50;
38/** How many failed jobs to pull per queue page before correlating. */
39const FAILED_SCAN = 200;
40 
41@Injectable()
42export class OrgNotificationsService {
43  constructor(
44    @Inject(PrismaService) private readonly prisma: PrismaService,
45    @Inject(RedisService) private readonly redis: RedisService
46  ) {}
47 
48  async health(orgId: string, req: Request) {
49    const userId = requireUser(req);
50    await assertOrgAdmin(this.prisma, orgId, userId);
51 
52    const workspaces = await this.prisma.workspace.findMany({
53      where: { organizationId: orgId },
54      select: { id: true }
55    });
56    const workspaceIds = workspaces.map((w) => w.id);
57 
58    // Bell volume (the IN_APP channel is the one the API writes).
59    const [inAppTotal, inAppUnread] = await Promise.all([
60      this.prisma.notification.count({ where: { workspaceId: { in: workspaceIds }, channel: "IN_APP" } }),
61      this.prisma.notification.count({
62        where: { workspaceId: { in: workspaceIds }, channel: "IN_APP", readAt: null }
63      })
64    ]);
65 
66    // Email-channel jobs: correlate the queue's email jobs (jobId
67    // `email_{notificationId}` — colon-free; the old `email:{id}` was rejected
68    // by BullMQ 5.x) to THIS org's notifications. The correlation
69    // is done in Postgres (org-scoped), never by trusting the queue.
70    const notificationIds = await this.prisma.notification.findMany({
71      where: { workspaceId: { in: workspaceIds } },
72      select: { id: true }
73    });
74    const orgNotificationIds = new Set(notificationIds.map((n) => n.id));
75 
76    // The email queue is the only one that matters here. Correlation is done
77    // by the job's PAYLOAD (`data.notificationIds`) — the field the worker's
78    // `buildInstantEmailData` + the digest enqueue set (and `emailJobSchema`
79    // reads to stamp `emailSentAt`) — NOT by the BullMQ job id. The instant
80    // key is now colon-free (`email_{id}`; the old `email:{id}` was rejected
81    // by BullMQ 5.x — see the T-178 conflict note, fixed 2026-10-12), but
82    // payload correlation stays the robust path: a digest job carries MANY
83    // ids and other email jobs (verify/invite/admin) carry none. It is the
84    // no-leak guarantee (org-scoped in Postgres).
85    const q = await this.openEmailQueue();
86    let emailQueued = 0;
87    let emailFailed = 0;
88    const failedJobs: Array<Record<string, unknown>> = [];
89    try {
90      const waiting = await q.getJobCountByTypes("waiting", "delayed");
91      const active = await q.getJobCountByTypes("active");
92      emailQueued = waiting + active;
93 
94      // Correlate: pull the queue's failed jobs, keep only those whose payload
95      // maps to a notification in THIS org. The payload carries `notificationIds`
96      // (an array — one for an instant email, many for a digest).
97      const failedAll = await q.getFailed(0, FAILED_SCAN - 1);
98      for (const j of failedAll) {
99        const ids = notificationIdsFromPayload(j.data);
100        const matched = ids.find((id) => orgNotificationIds.has(id));
101        if (!matched) continue;
102        emailFailed++;
103        if (failedJobs.length < FAILED_LIST_CAP) {
104          failedJobs.push({
105            notificationId: matched,
106            jobId: j.id,
107            name: j.name,
108            failedReason: j.failedReason ? String(j.failedReason).slice(0, 500) : null,
109            failedAt: j.finishedOn ? new Date(j.finishedOn).toISOString() : null,
110            attemptsMade: j.attemptsMade
111          });
112        }
113      }
114      failedJobs.sort((a, b) => String(b.failedAt).localeCompare(String(a.failedAt)));
115    } finally {
116      await q.close();
117    }
118 
119    return {
120      workspaces: workspaceIds.length,
121      inAppTotal,
122      inAppUnread,
123      emailQueued,
124      emailFailed,
125      failedJobs
126    };
127  }
128 
129  private async openEmailQueue() {
130    const { Queue } = await import("bullmq");
131    return new Queue("email", { connection: this.redis.getRedis().options as never });
132  }
133}
134 
135/** Extract the notification ids from an email job's payload. The worker's
136 *  `buildInstantEmailData` (instant) and the digest enqueue both set
137 *  `notificationIds: string[]` (one for instant, many for digest) — the field
138 *  `emailJobSchema` reads to stamp `emailSentAt`. Returns [] for non-
139 *  notification payloads (verify / invite / admin alerts omit it). */
140function notificationIdsFromPayload(data: unknown): string[] {
141  if (!data || typeof data !== "object") return [];
142  const ids = (data as Record<string, unknown>).notificationIds;
143  if (!Array.isArray(ids)) return [];
144  return ids.filter((id): id is string => typeof id === "string" && id.length > 0);
145}
146