jobs.service.ts

jobs.service.tsapps/api/src/modules/org/jobs.service.tsraw ← org6,975 B← Back to the app
1/**
2 * T-176 — org-admin JOBS tab (background job status + failures).
3 *
4 * BullMQ is CLUSTER-wide (every org shares the seven worker queues), so this
5 * tab reports cluster queue state — the org-admin gate is the access control
6 * (an org admin is a trusted platform operator; the response says
7 * `scope: "cluster"` so the UI never pretends the numbers are org-scoped).
8 *
9 * Surface:
10 *   GET  /orgs/:orgId/admin/jobs            — per-queue state counts
11 *   GET  /orgs/:orgId/admin/jobs/failed     — recent failed jobs (the DLQ view)
12 *   POST /orgs/:orgId/admin/jobs/requeue    — re-run ONE failed job (audited)
13 *
14 * The queue name is a server-side enum (never caller-supplied string →
15 * `new Queue(<untrusted>)`). Requeue resets the job's attempt counter and
16 * returns it to waiting (BullMQ's `Job.retry`) — the standard DLQ re-run.
17 */
18import { Inject, Injectable } from "@nestjs/common";
19import { z } from "zod";
20import type { Request } from "express";
21import { PrismaService } from "../../common/prisma.service.js";
22import { RedisService } from "../../common/redis.service.js";
23import { AuditService } from "../../common/audit.service.js";
24import { ApiError } from "../../common/errors.js";
25import { requireUser } from "../../common/types.js";
26import { assertOrgAdmin } from "./guards.js";
27 
28/**
29 * The seven worker queues (mirrors `apps/worker/src/queues.ts::QUEUES` — kept
30 * as a literal here so the API does not import the worker package; the names
31 * are a stable contract, and a name added to the worker without adding it here
32 * simply stays hidden from this tab).
33 */
34const QUEUE_NAMES = ["files", "search-index", "webhooks", "automations", "email", "ai", "maintenance"] as const;
35export type AdminQueueName = (typeof QUEUE_NAMES)[number];
36 
37/** The failed-jobs list cap (one page of the DLQ view). */
38const FAILED_LIST_CAP = 100;
39 
40/** How much of a job's payload to surface (DLQ payloads can be big). */
41const PAYLOAD_PREVIEW_CHARS = 2000;
42 
43const RequeueBody = z.object({
44  queue: z.enum(QUEUE_NAMES),
45  jobId: z.string().min(1).max(128)
46});
47 
48@Injectable()
49export class OrgJobsService {
50  constructor(
51    @Inject(PrismaService) private readonly prisma: PrismaService,
52    @Inject(RedisService) private readonly redis: RedisService,
53    @Inject(AuditService) private readonly audit: AuditService
54  ) {}
55 
56  /** Per-queue state counts (waiting/active/completed/failed). */
57  async overview(orgId: string, req: Request) {
58    const userId = requireUser(req);
59    await assertOrgAdmin(this.prisma, orgId, userId);
60    const queues: Array<{ name: AdminQueueName; waiting: number; active: number; completed: number; failed: number }> = [];
61    // One queue handle at a time (the house pattern — the API never keeps
62    // long-lived BullMQ connections; each handle is closed after use).
63    for (const name of QUEUE_NAMES) {
64      const q = await this.openQueue(name);
65      try {
66        const [waiting, active, completed, failed] = await Promise.all([
67          q.getJobCountByTypes("waiting", "delayed"),
68          q.getJobCountByTypes("active"),
69          q.getJobCountByTypes("completed"),
70          q.getJobCountByTypes("failed")
71        ]);
72        queues.push({ name, waiting, active, completed, failed });
73      } finally {
74        await q.close();
75      }
76    }
77    return {
78      // BullMQ is cluster-wide (all orgs share the worker queues) — the
79      // org-admin gate is the access control; the label keeps the UI honest.
80      scope: "cluster" as const,
81      queues
82    };
83  }
84 
85  /** Recent failed jobs (the DLQ view) — all queues or one. */
86  async failed(
87    orgId: string,
88    req: Request,
89    queue?: string,
90    limit = 20
91  ): Promise<{ scope: "cluster"; jobs: Array<Record<string, unknown>> }> {
92    const userId = requireUser(req);
93    await assertOrgAdmin(this.prisma, orgId, userId);
94    const cap = Math.max(1, Math.min(limit, FAILED_LIST_CAP));
95    const names = queue ? [queue as AdminQueueName] : [...QUEUE_NAMES];
96    const jobs: Array<Record<string, unknown>> = [];
97    for (const name of names) {
98      const q = await this.openQueue(name);
99      try {
100        const failed = await q.getFailed(0, cap - 1);
101        for (const j of failed) {
102          jobs.push(this.jobView(name, j));
103        }
104      } finally {
105        await q.close();
106      }
107    }
108    jobs.sort((a, b) => String(b.failedAt).localeCompare(String(a.failedAt)));
109    return { scope: "cluster", jobs: jobs.slice(0, cap) };
110  }
111 
112  /**
113   * Re-run ONE failed job (resets its attempt counter, back to waiting).
114   * The job must actually be in the failed state — requeueing a live job
115   * would be a no-op surprise, so it is a 409 instead.
116   */
117  async requeue(
118    orgId: string,
119    req: Request,
120    body: { queue: AdminQueueName; jobId: string }
121  ): Promise<{ queued: true; queue: AdminQueueName; jobId: string }> {
122    const userId = requireUser(req);
123    await assertOrgAdmin(this.prisma, orgId, userId);
124    const q = await this.openQueue(body.queue);
125    try {
126      const job = await q.getJob(body.jobId);
127      if (!job) {
128        // No-leak: a job id from another queue (or a vanished id) is a 404.
129        throw ApiError.notFound("Job not found in this queue.");
130      }
131      const state = await job.getState();
132      if (state !== "failed") {
133        throw ApiError.conflict("Only failed jobs can be requeued.", [
134          { reason: "job_not_failed", state }
135        ]);
136      }
137      await job.retry();
138      await this.audit.audit(req, req.auth, {
139        action: "jobs.failed_job_requeued",
140        resourceType: "queue",
141        resourceId: body.queue,
142        metadata: { orgId, queue: body.queue, jobId: body.jobId, jobName: job.name }
143      });
144      return { queued: true, queue: body.queue, jobId: body.jobId };
145    } finally {
146      await q.close();
147    }
148  }
149 
150  /** One failed-job row for the DLQ view (payload truncated). */
151  private jobView(name: AdminQueueName, j: {
152    id?: string;
153    name: string;
154    data?: unknown;
155    failedReason?: string | null;
156    /** BullMQ surfaces the failure timestamp as `finishedOn` (ms since epoch). */
157    finishedOn?: number;
158    attemptsMade: number;
159  }) {
160    const payload = (() => {
161      try {
162        const s = JSON.stringify(j.data ?? null);
163        return s.length > PAYLOAD_PREVIEW_CHARS ? `${s.slice(0, PAYLOAD_PREVIEW_CHARS)}…` : s;
164      } catch {
165        return null;
166      }
167    })();
168    return {
169      queue: name,
170      jobId: j.id ?? null,
171      name: j.name,
172      payload,
173      failedReason: j.failedReason ? String(j.failedReason).slice(0, 500) : null,
174      // BullMQ surfaces the failure timestamp as `finishedOn` (ms since epoch).
175      failedAt: j.finishedOn ? new Date(j.finishedOn).toISOString() : null,
176      attemptsMade: j.attemptsMade
177    };
178  }
179 
180  private async openQueue(name: AdminQueueName) {
181    const { Queue } = await import("bullmq");
182    return new Queue(name, { connection: this.redis.getRedis().options as never });
183  }
184}
185