audit-export-s3.integration.test.ts

audit-export-s3.integration.test.tstests/integration/audit-export-s3.integration.test.tsraw ← integration11,703 B← Back to the app
1/**
2 * TASK-0133 — the async (>10k) audit CSV EXPORT, the two halves of the
3 * `> SYNC_ROW_LIMIT (10_000)` path:
4 *
5 *   PART 1 (always runs, no S3 network call):
6 *     - the API switches on the matching row count: >10k → `202 {exportId,
7 *       cap, total}` + an enqueued `maintenance`/`audit.export` job. The 202
8 *       IS the enqueue assertion (the controller `await`s `enqueueExportJob`
9 *       before responding).
10 *     - the worker flips the `AuditExport` row to `ready` (simulated here by
11 *       updating the row — the real S3 write is Part 2).
12 *     - `GET .../audit/exports/:id` mints a SINGLE-USE 15-minute presigned
13 *       GET. This is PURE SIGNING (no S3 round trip), so it runs without
14 *       MinIO: the first call returns the URL + flips `downloadUsed`; every
15 *       subsequent call (with or without `?redownload=1`) returns no URL.
16 *
17 *   PART 2 (`describe.skipIf(!minioReady)` — the real S3 leg):
18 *     - calls the worker's `runAuditExport` directly (the same function the
19 *       `maintenance` processor invokes) against real Postgres + a real
20 *       S3/MinIO: it streams the org-scoped rows, writes the 13-column CSV to
21 *       `exports/{orgId}/{exportId}.csv`, flips the row to `ready`, and
22 *       emits `audit.export.completed`. We then read the object back from S3
23 *       (content-type + header + row count) and assert the audit row.
24 *
25 * Truncation at the 500k cap is UNIT-tested (the cap is a non-injectable
26 * constant; seeding 500k rows is not a realistic integration fixture).
27 *
28 * Org scoping: `AuditLog` has NO `organizationId` column — the org scope is
29 * `workspaceId IN (org's workspaces)` OR (org-level rows). We therefore create
30 * the workspace with `organizationId = orgId` and seed rows against it
31 * (`@factory/shared/events/audit-scope` is the single source of truth).
32 *
33 * NOTE (integration fetch bodies): `app.request` returns a `Response` whose
34 * body is a read-once stream. Every call is funneled through `read()`, which
35 * consumes the body ONCE (raw text + a JSON parse) so an assertion message can
36 * quote the raw body without a second read.
37 */
38import { afterAll, beforeAll, describe, expect, it } from "vitest";
39import { PrismaClient } from "@prisma/client";
40import { Redis } from "ioredis";
41import { createTestApp, type TestAppHandle } from "./helpers/api";
42import { authHeaders, createUser } from "./helpers/seed";
43import { getRedisUrl } from "./helpers/containers";
44import { runAuditExport, ASYNC_ROW_CAP } from "../../apps/worker/src/processors/audit-export.js";
45import { resetEnvCache } from "../../apps/worker/src/config.js";
46import { S3Client } from "../../apps/worker/src/lib/s3.js";
47import { AUDIT_CSV_HEADER } from "@factory/shared";
48 
49// ── MinIO gate (Part 2) ─────────────────────────────────────────────────
50const S3_ENDPOINT = (process.env.S3_ENDPOINT ?? "http://localhost:9000").replace(/\/+$/, "");
51const S3_REGION = process.env.S3_REGION ?? "us-east-1";
52const S3_BUCKET = process.env.S3_BUCKET_FILES ?? process.env.S3_BUCKET ?? "factory-files";
53const S3_ACCESS = process.env.S3_ACCESS_KEY ?? "minioadmin";
54const S3_SECRET = process.env.S3_SECRET_KEY ?? "minioadmin";
55 
56async function minioUp(url: string): Promise<boolean> {
57  try {
58    const res = await fetch(`${url}/minio/health/live`, { signal: AbortSignal.timeout(4_000) });
59    return res.ok;
60  } catch {
61    return false;
62  }
63}
64const minioReady = await minioUp(S3_ENDPOINT);
65 
66/** Read a response body ONCE (raw text + a JSON parse for the convenience). */
67async function read(res: Response | Promise<Response>): Promise<{ status: number; raw: string; data: Record<string, unknown> }> {
68  const r = await res;
69  const raw = await r.text();
70  let data: Record<string, unknown> = {};
71  try {
72    data = JSON.parse(raw) as Record<string, unknown>;
73  } catch {
74    /* not JSON (e.g. text/csv) */
75  }
76  return { status: r.status, raw, data };
77}
78 
79let app: TestAppHandle;
80let redis: Redis;
81let prisma: PrismaClient;
82 
83/** Create an org through the API; the caller becomes the OWNER member. */
84async function makeOrg(accessToken: string): Promise<string> {
85  const out = await read(
86    app.request("POST", "/api/v1/orgs", {
87      headers: authHeaders(accessToken),
88      body: { name: `AsyncExport Org ${Date.now()}-${Math.random().toString(36).slice(2, 7)}` }
89    })
90  );
91  const orgId = out.data.id as string | undefined;
92  if (!orgId) throw new Error(`makeOrg failed: ${out.status} ${out.raw}`);
93  return orgId;
94}
95 
96/** Seed N audit rows into the org (via an org-owned workspace) + return the ws. */
97async function seedRows(orgId: string, ownerId: string, n: number): Promise<string> {
98  const ws = await prisma.workspace.create({
99    data: { name: "ws", slug: `ae-ws-${Date.now()}-${Math.random().toString(36).slice(2, 7)}`, organizationId: orgId, ownerId }
100  });
101  const rows = Array.from({ length: n }, (_, i) => ({
102    actorType: "USER" as const,
103    actorId: ownerId,
104    workspaceId: ws.id,
105    resourceType: "page",
106    resourceId: `page_${i}`,
107    action: i % 2 === 0 ? "page.created" : "page.updated",
108    metadata: {}
109  }));
110  // Batches keep each single insert statement small (createMany is one insert).
111  for (let i = 0; i < rows.length; i += 2_000) {
112    await prisma.auditLog.createMany({ data: rows.slice(i, i + 2_000) as never });
113  }
114  return ws.id;
115}
116 
117/** Insert a processing AuditExport row (the API's async branch shape). */
118async function seedExport(orgId: string, total: number): Promise<string> {
119  const exportId = `ae-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
120  await prisma.auditExport.create({
121    data: { id: exportId, organizationId: orgId, status: "processing", totalRows: total, capRows: ASYNC_ROW_CAP, filters: JSON.stringify({}) as never }
122  });
123  return exportId;
124}
125 
126describe("TASK-0133 async audit export (the >10k path)", () => {
127  beforeAll(async () => {
128    app = await createTestApp();
129    prisma = app.prisma;
130    redis = new Redis(await getRedisUrl(), { maxRetriesPerRequest: null });
131  }, 180_000);
132  afterAll(async () => {
133    await redis?.quit().catch(() => {});
134    await app?.close();
135  });
136 
137  it("PART 1: API >10k → 202 {exportId,cap,total} (the enqueue), then a SINGLE-USE presigned URL (pure signing, no S3)", async () => {
138    const owner = await createUser(app.request);
139    const orgId = await makeOrg(owner.accessToken!); // also emits an org.created row
140    // 10,001 workspace rows — just over SYNC_ROW_LIMIT (10_000) → the async
141    // branch. (+ the org.created org-level row; the exact total is >10_000.)
142    await seedRows(orgId, owner.userId, 10_001);
143 
144    const h = authHeaders(owner.accessToken!);
145    const startRes = await read(app.request("POST", `/api/v1/orgs/${orgId}/admin/audit/export`, { headers: h, body: {} }));
146    expect(startRes.status, startRes.raw).toBe(202);
147    const start = startRes.data as { exportId: string; cap: number; total: number };
148    expect(start.total).toBeGreaterThan(10_000);
149    expect(start.cap).toBe(ASYNC_ROW_CAP);
150    expect(start.exportId).toMatch(/^[A-Za-z0-9-]+$/);
151    // The 202 only happens AFTER `await enqueueExportJob(...)` — so a 202 IS
152    // the proof the `maintenance`/`audit.export` job was enqueued.
153 
154    // The row starts `processing`; no URL until it is ready.
155    const pending = (await read(app.request("GET", `/api/v1/orgs/${orgId}/admin/audit/exports/${start.exportId}`, { headers: h }))).data as {
156      status: string;
157      downloadUrl?: string
158    };
159    expect(pending.status).toBe("processing");
160    expect(pending.downloadUrl).toBeUndefined();
161 
162    // Simulate the worker finishing (the real S3 write is PART 2): flip the
163    // row to `ready` with the row count.
164    await prisma.auditExport.update({
165      where: { id: start.exportId },
166      data: { status: "ready", rows: 10_001, truncated: false, completedAt: new Date() }
167    });
168 
169    // FIRST status call → mints the single-use URL (pure SigV4 signing).
170    const first = (await read(app.request("GET", `/api/v1/orgs/${orgId}/admin/audit/exports/${start.exportId}`, { headers: h }))).data as {
171      status: string;
172      rows: number;
173      truncated: boolean;
174      downloadUrl?: string;
175      urlTtlSec?: number
176    };
177    expect(first.status).toBe("ready");
178    expect(first.rows).toBe(10_001);
179    expect(first.truncated).toBe(false);
180    expect(first.downloadUrl).toBeDefined();
181    expect(first.urlTtlSec).toBe(900);
182    // The signed URL targets the deterministic object key + carries SigV4 sig.
183    expect(first.downloadUrl).toContain(`exports/${orgId}/${start.exportId}.csv`);
184    expect(first.downloadUrl).toMatch(/X-Amz-Signature=/);
185 
186    // SINGLE-USE: the second call (and even `?redownload=1`) returns NO url —
187    // `downloadUsed` was flipped on the first mint.
188    const second = (await read(app.request("GET", `/api/v1/orgs/${orgId}/admin/audit/exports/${start.exportId}`, { headers: h }))).data as {
189      downloadUrl?: string
190    };
191    expect(second.downloadUrl).toBeUndefined();
192    const third = (await read(
193      app.request("GET", `/api/v1/orgs/${orgId}/admin/audit/exports/${start.exportId}?redownload=1`, { headers: h })
194    )).data as { downloadUrl?: string };
195    expect(third.downloadUrl).toBeUndefined();
196  });
197 
198  describe.skipIf(!minioReady)("PART 2: worker runAuditExport writes the CSV to S3 (MinIO)", () => {
199    it("streams the org-scoped rows → putObject → headObject/getObjectStream round trip + audit.export.completed", async () => {
200      // Reset the worker's cached env so it parses the CURRENT process.env
201      // (the test app set DATABASE_URL/REDIS_URL; the S3_* are at their
202      // defaults = this MinIO endpoint).
203      resetEnvCache();
204 
205      const owner = await createUser(app.request);
206      const orgId = await makeOrg(owner.accessToken!); // emits the org.created org-level row
207      const N = 120;
208      await seedRows(orgId, owner.userId, N);
209      const exportId = await seedExport(orgId, N);
210 
211      const res = await runAuditExport(prisma, redis, orgId, exportId);
212      expect(res.status).toBe("ready");
213      // The org scope = the N workspace rows + the org.created org-level row
214      // (the OR-branch), so at least N (and in practice N+1) rows flow through.
215      expect(res.rows).toBeGreaterThanOrEqual(N);
216      expect(res.truncated).toBe(false);
217      expect(res.key).toBe(`exports/${orgId}/${exportId}.csv`);
218 
219      // The DB row is ready with the same count.
220      const row = await prisma.auditExport.findUnique({ where: { id: exportId } });
221      expect(row?.status).toBe("ready");
222      expect(row?.rows).toBe(res.rows);
223 
224      // The object is in S3 with the right content type + 13-column header, and
225      // the CSV invariant: exactly one line per counted row (+ header).
226      const s3 = new S3Client({ endpoint: S3_ENDPOINT, region: S3_REGION, bucket: S3_BUCKET, accessKey: S3_ACCESS, secretKey: S3_SECRET });
227      const head = await s3.headObject(res.key as string);
228      expect(head).not.toBeNull();
229      expect(head?.contentType).toContain("text/csv");
230      const buf = await s3.getObjectStream(res.key as string);
231      const csv = buf.toString("utf8");
232      const lines = csv.trim().split("\n");
233      expect(lines[0]).toBe(AUDIT_CSV_HEADER);
234      expect(lines.length).toBe(res.rows + 1); // header + res.rows rows
235 
236      // The completion audit row (actor SYSTEM, action audit.export.completed).
237      const audit = await prisma.auditLog.findFirst({
238        where: { action: "audit.export.completed", resourceId: orgId, metadata: { path: ["exportId"], equals: exportId } }
239      });
240      expect(audit).not.toBeNull();
241    });
242  });
243});
244