audit.service.ts

audit.service.tsapps/api/src/modules/org/audit.service.tsraw ← org14,447 B← Back to the app
1/**
2 * TASK-0133 — org-scoped AUDIT VIEWER service.
3 *
4 * Org scoping (recorded decision — no new index; EXPLAIN on the live 12k-row
5 * table uses `audit_logs_workspaceId_action_createdAt_idx`): the org's rows
6 * are
7 *   - rows in ANY of the org's workspaces (`workspaceId IN wsIds`), PLUS
8 *   - ORG-LEVEL rows: `workspaceId IS NULL` and (metadata.orgId = orgId —
9 *     the SSO/SCIM admin + login emitters) OR (resourceType
10 *     "organization"|"org" and resourceId = orgId — org.created/org.updated/
11 *     SSO-admin + my own org.policy.updated/org.member_updated).
12 *
13 * Read-only by construction: no UPDATE/DELETE path exists on AuditLog (the
14 * audit data is immutable — the router-walk test in the integration suite
15 * asserts the admin surface exposes no mutation routes).
16 *
17 * CSV export (the stable 13-column contract in `audit-csv.ts`):
18 *   - ≤10k matching rows → SYNCHRONOUS 200 (streamed text/csv)
19 *   - >10k               → 202 {exportId}: a `maintenance` queue job
20 *     (`audit.export`) streams cursor batches (5k) into S3
21 *     `exports/{orgId}/{exportId}.csv` (retained 7 d), capped at 500k rows
22 *     (`truncated:true`). `GET .../exports/:id` → status + a SINGLE-USE,
23 *     15-minute presigned download (the flag is flipped when the URL is
24 *     issued; a second poll without a download returns the status again
25 *     after the TTL, and an explicit re-request via `?redownload=1` mints a
26 *     fresh one). One active export per org (409 otherwise).
27 */
28import { Inject, Injectable } from "@nestjs/common";
29import type { Prisma } from "@prisma/client";
30import { randomUUID } from "node:crypto";
31// The API compiles under node10 moduleResolution → root barrel only (the
32// shared package re-exports the TASK-0133 audit surface at the root).
33import {
34  AUDIT_CATALOG_NAMES,
35  auditOrgScopeWhere,
36  auditScopedWhere,
37  renderAuditCsv,
38  AUDIT_CSV_HEADER,
39  type AuditFilters as SharedAuditFilters,
40  type AuditCsvRow
41} from "@factory/shared";
42import type { Request } from "express";
43import { ApiError } from "../../common/errors.js";
44import { jobTraceMetadata } from "../../common/trace-metadata.js";
45import { PrismaService } from "../../common/prisma.service.js";
46import { RedisService } from "../../common/redis.service.js";
47import { AuditService } from "../../common/audit.service.js";
48import { requireUser } from "../../common/types.js";
49import { assertOrgAdmin } from "./guards.js";
50 
51const DAY_MS = 86_400_000;
52const SYNC_ROW_LIMIT = 10_000;
53const ASYNC_ROW_CAP = 500_000;
54const EXPORT_TTL_SEC = 7 * 86_400;
55 
56export type ExportStatus = "processing" | "ready" | "failed";
57 
58// ────────────────────────────── query core ─────────────────────────────────
59// The org scope + filter builders live in `@factory/shared/events/audit-scope`
60// (single source shared with the worker's retention purge + async export).
61 
62export type AuditFilters = SharedAuditFilters;
63 
64export interface ParsedAuditQuery extends AuditFilters {
65  page: number;
66  pageSize: number;
67}
68 
69/**
70 * Parse + validate the viewer query. `from`/`to` are ISO date-strings (or
71 * Dates); the window may not exceed 90 days (400). `pageSize` ≤ 100, default
72 * 50. Throws ApiError.validation on bad input.
73 */
74export function parseAuditQuery(q: Record<string, unknown>): ParsedAuditQuery {
75  const str = (k: string): string | undefined => {
76    const v = q[k];
77    if (v === undefined || v === null || v === "") return undefined;
78    if (typeof v !== "string") throw ApiError.validation(`Invalid ${k}.`, [{ field: k, issue: "not_a_string" }]);
79    return v;
80  };
81  let from: Date | undefined;
82  let to: Date | undefined;
83  const fromRaw = str("from");
84  const toRaw = str("to");
85  if (fromRaw) {
86    from = new Date(fromRaw);
87    if (Number.isNaN(from.getTime())) throw ApiError.validation("Invalid from date.", [{ field: "from", issue: "bad_date" }]);
88  }
89  if (toRaw) {
90    to = new Date(toRaw);
91    if (Number.isNaN(to.getTime())) throw ApiError.validation("Invalid to date.", [{ field: "to", issue: "bad_date" }]);
92  }
93  if (from && to && to.getTime() - from.getTime() > 90 * DAY_MS) {
94    throw ApiError.validation("Date range is limited to 90 days per query.", [{ reason: "date_range_too_wide" }]);
95  }
96  if (from && to && to < from) {
97    throw ApiError.validation("to must be after from.", [{ reason: "date_range_inverted" }]);
98  }
99  const page = q.page === undefined ? 1 : Number(q.page);
100  const pageSize = q.pageSize === undefined ? 50 : Number(q.pageSize);
101  if (!Number.isInteger(page) || page < 1) throw ApiError.validation("Invalid page.", [{ field: "page", issue: "bad_page" }]);
102  if (!Number.isInteger(pageSize) || pageSize < 1 || pageSize > 100) {
103    throw ApiError.validation("pageSize must be 1..100.", [{ field: "pageSize", issue: "bad_pageSize" }]);
104  }
105  return {
106    actorId: str("actorId"),
107    actorType: str("actorType"),
108    action: str("action"),
109    actionPrefix: str("actionPrefix"),
110    resourceType: str("resourceType"),
111    workspaceId: str("workspaceId"),
112    from,
113    to,
114    page,
115    pageSize
116  };
117}
118 
119@Injectable()
120export class OrgAuditService {
121  constructor(
122    @Inject(PrismaService) private readonly prisma: PrismaService,
123    @Inject(RedisService) private readonly redis: RedisService,
124    @Inject(AuditService) private readonly audit: AuditService
125  ) {}
126 
127  /** The org-admin gate (every audit-viewer route is admin-only). */
128  private async requireAdmin(orgId: string, req: Request): Promise<void> {
129    const userId = requireUser(req);
130    await assertOrgAdmin(this.prisma, orgId, userId);
131  }
132 
133  private async wsIds(orgId: string): Promise<string[]> {
134    const rows = await this.prisma.workspace.findMany({
135      where: { organizationId: orgId },
136      select: { id: true }
137    });
138    return rows.map((r) => r.id);
139  }
140 
141  /** The full filter set (org scope + user filters) for a query. */
142  private async scope(orgId: string, f: AuditFilters): Promise<Prisma.AuditLogWhereInput> {
143    const ids = await this.wsIds(orgId);
144    return auditScopedWhere(orgId, ids, f) as Prisma.AuditLogWhereInput;
145  }
146 
147  /** `GET /orgs/:orgId/admin/audit` — paginated + the deep-link existence map. */
148  async list(orgId: string, rawQuery: Record<string, unknown>, req: Request) {
149    await this.requireAdmin(orgId, req);
150    const q = parseAuditQuery(rawQuery);
151    const where = await this.scope(orgId, q);
152    const [total, rows] = await Promise.all([
153      this.prisma.auditLog.count({ where }),
154      this.prisma.auditLog.findMany({
155        where,
156        orderBy: [{ createdAt: "desc" }, { id: "desc" }],
157        skip: (q.page - 1) * q.pageSize,
158        take: q.pageSize
159      })
160    ]);
161    // Deep-link existence for this page's linkable resources (pages,
162    // databases, records) — one bulk query per type (the drawer deep-links
163    // only when the resource still exists; gone → plain id).
164    const linkTypes: Record<string, string> = {};
165    const idsFor = (type: string) =>
166      [...new Set(rows.filter((r) => r.resourceType === type && r.resourceId).map((r) => r.resourceId))];
167    const [pageIds, dbIds, recIds] = await Promise.all([
168      this.prisma.page.findMany({ where: { id: { in: idsFor("page") } }, select: { id: true } }),
169      this.prisma.database.findMany({ where: { id: { in: idsFor("database") } }, select: { id: true } }),
170      this.prisma.databaseRecord.findMany({ where: { id: { in: idsFor("record") } }, select: { id: true } })
171    ]);
172    for (const p of pageIds) linkTypes[p.id] = "page";
173    for (const d of dbIds) linkTypes[d.id] = "database";
174    for (const r of recIds) linkTypes[r.id] = "record";
175 
176    return {
177      data: rows.map((r) => ({
178        ...r,
179        resourceLink: r.resourceId && linkTypes[r.resourceId] ? { type: linkTypes[r.resourceId] } : null
180      })),
181      total,
182      page: q.page,
183      pageSize: q.pageSize
184    };
185  }
186 
187  /** `GET /orgs/:orgId/admin/audit/actions` — the autocomplete list. */
188  async distinctActions(orgId: string, req: Request) {
189    await this.requireAdmin(orgId, req);
190    const wsIds = await this.wsIds(orgId);
191    const known = new Set(AUDIT_CATALOG_NAMES as readonly string[]);
192    const rows = await this.prisma.auditLog.findMany({
193      where: auditOrgScopeWhere(orgId, wsIds) as Prisma.AuditLogWhereInput,
194      select: { action: true },
195      distinct: ["action"],
196      take: 1000
197    });
198    const actions = [...new Set(rows.map((r) => r.action))].sort();
199    // Catalog actions first (the expected surface), then anything observed.
200    const ordered = [...actions.filter((a) => known.has(a)), ...actions.filter((a) => !known.has(a))];
201    return { data: ordered };
202  }
203 
204  // ────────────────────────────── CSV export ───────────────────────────────
205 
206  /** The S3 object key for an export (the worker writes to the SAME key). */
207  private exportKey(orgId: string, exportId: string): string {
208    return `exports/${orgId}/${exportId}.csv`;
209  }
210 
211  private async nameMaps(rows: AuditCsvRow[]): Promise<{
212    users: Map<string, string>;
213    workspaces: Map<string, string>;
214  }> {
215    const userIds = [...new Set(rows.filter((r) => r.actorType === "USER").map((r) => r.actorId))];
216    const wsIds = [...new Set(rows.map((r) => r.workspaceId).filter((w): w is string => Boolean(w)))];
217    const [users, workspaces] = await Promise.all([
218      this.prisma.user.findMany({ where: { id: { in: userIds } }, select: { id: true, name: true } }),
219      this.prisma.workspace.findMany({ where: { id: { in: wsIds } }, select: { id: true, name: true } })
220    ]);
221    return {
222      users: new Map(users.map((u) => [u.id, u.name])),
223      workspaces: new Map(workspaces.map((w) => [w.id, w.name]))
224    };
225  }
226 
227  async countForExport(orgId: string, f: AuditFilters): Promise<number> {
228    return this.prisma.auditLog.count({ where: await this.scope(orgId, f) });
229  }
230 
231  /**
232   * `POST /orgs/:orgId/admin/audit/export`. Returns either the inline CSV
233   * (sync, ≤10k rows) or `202 {exportId}` (async). The caller sets the
234   * response shape (the controller inspects `kind`).
235   */
236  async startExport(orgId: string, rawQuery: Record<string, unknown>, req: Request) {
237    await this.requireAdmin(orgId, req);
238    const q = parseAuditQuery(rawQuery);
239    const where = await this.scope(orgId, q);
240    const total = await this.prisma.auditLog.count({ where });
241    if (total <= SYNC_ROW_LIMIT) {
242      const rows = await this.prisma.auditLog.findMany({
243        where,
244        orderBy: [{ createdAt: "desc" }, { id: "desc" }],
245        take: SYNC_ROW_LIMIT
246      });
247      const csvRows: AuditCsvRow[] = rows.map((r) => ({
248        id: r.id,
249        createdAt: r.createdAt,
250        actorType: r.actorType,
251        actorId: r.actorId,
252        workspaceId: r.workspaceId,
253        resourceType: r.resourceType,
254        resourceId: r.resourceId,
255        action: r.action,
256        ipAddress: r.ipAddress,
257        userAgent: r.userAgent,
258        metadata: r.metadata
259      }));
260      const maps = await this.nameMaps(csvRows);
261      return { kind: "sync" as const, total, csv: renderAuditCsv(csvRows, maps.users, maps.workspaces) };
262    }
263    // Async: one active export per org.
264    const active = await this.prisma.auditExport.findFirst({
265      where: { organizationId: orgId, status: "processing" },
266      select: { id: true }
267    });
268    if (active) {
269      throw ApiError.conflict("An export is already running for this organization.", [
270        { reason: "export_in_progress", exportId: active.id }
271      ]);
272    }
273    const exportId = randomUUID();
274    await this.prisma.auditExport.create({
275      data: {
276        id: exportId,
277        organizationId: orgId,
278        status: "processing",
279        filters: JSON.stringify(q) as unknown as object,
280        totalRows: total,
281        capRows: ASYNC_ROW_CAP
282      }
283    });
284    return { kind: "async" as const, total, exportId, cap: ASYNC_ROW_CAP };
285  }
286 
287  /** Enqueue the worker `audit.export` job (maintenance queue). */
288  async enqueueExportJob(exportId: string, orgId: string): Promise<void> {
289    const { Queue } = await import("bullmq");
290    const q = new Queue("maintenance", {
291      connection: this.redis.getRedis().options as never
292    });
293    try {
294      await q.add("maintenance", { type: "audit.export", orgId, exportId }, { jobId: exportId, ...jobTraceMetadata() });
295    } finally {
296      await q.close();
297    }
298  }
299 
300  /** `GET /orgs/:orgId/admin/audit/exports/:exportId` — status + single-use URL. */
301  async exportStatus(orgId: string, exportId: string, redownload: boolean, req: Request) {
302    await this.requireAdmin(orgId, req);
303    const row = await this.prisma.auditExport.findFirst({
304      where: { id: exportId, organizationId: orgId }
305    });
306    if (!row) throw ApiError.notFound("Export not found.");
307    const payload: Record<string, unknown> = {
308      exportId: row.id,
309      status: row.status.toLowerCase(),
310      rows: row.rows,
311      truncated: row.truncated,
312      createdAt: row.createdAt,
313      completedAt: row.completedAt
314    };
315    if (row.error) payload.error = row.error;
316    if (row.status === "ready" && row.downloadUsed === false && (redownload || row.downloadUrlAt === null)) {
317      // Mint the single-use URL (15 min) + flip the flag.
318      const key = `exports/${orgId}/${row.id}.csv`;
319      const url = await this.presignCsv(orgId, key);
320      await this.prisma.auditExport.update({
321        where: { id: row.id },
322        data: { downloadUrlAt: new Date(), downloadUsed: true }
323      });
324      payload.downloadUrl = url;
325      payload.urlTtlSec = 900;
326    }
327    return payload;
328  }
329 
330  /**
331   * The S3 presign for a CSV (15 min, attachment disposition). Uses the API's
332   * files-bucket SigV4 signer (no network call — pure signing), matching the
333   * files module's bucket.
334   */
335  private async presignCsv(orgId: string, key: string): Promise<string> {
336    const { filesBucketConfig } = await import("../files/s3.js");
337    const { presignGet } = await import("@factory/storage");
338    const signed = presignGet({
339      config: filesBucketConfig(),
340      key,
341      expiresSec: 900,
342      now: new Date(),
343      responseContentDisposition: `attachment; filename="audit-${orgId.slice(0, 8)}.csv"`
344    });
345    return signed.url;
346  }
347}
348 
349export { AUDIT_CSV_HEADER, EXPORT_TTL_SEC };
350