1
26import type { PrismaClient, Prisma } from "@prisma/client";
27import type { Redis } from "ioredis";
28import { S3Client } from "../lib/s3.js";
29import { asJson } from "../lib/prisma.js";
30import { getLogger } from "../logger.js";
31import { loadEnv } from "../config.js";
32import { auditScopedWhere, type AuditFilters } from "@factory/shared/events/audit-scope";
33import { AUDIT_CSV_HEADER, auditCsvLine, type AuditCsvRow } from "@factory/shared/events/audit-csv";
34
35const log = getLogger("audit-export");
36
37const BATCH = 5_000;
38export const ASYNC_ROW_CAP = 500_000;
39
40
41function parseStoredFilters(raw: unknown): AuditFilters {
42 const f = (raw as Record<string, unknown> | null) ?? {};
43 const date = (v: unknown): Date | undefined =>
44 typeof v === "string" && v ? new Date(v) : undefined;
45 return {
46 actorId: typeof f.actorId === "string" ? f.actorId : undefined,
47 actorType: typeof f.actorType === "string" ? f.actorType : undefined,
48 action: typeof f.action === "string" ? f.action : undefined,
49 actionPrefix: typeof f.actionPrefix === "string" ? f.actionPrefix : undefined,
50 resourceType: typeof f.resourceType === "string" ? f.resourceType : undefined,
51 workspaceId: typeof f.workspaceId === "string" ? f.workspaceId : undefined,
52 from: date(f.from),
53 to: date(f.to)
54 };
55}
56
57
62export async function runAuditExport(
63 prisma: PrismaClient,
64 redis: Redis,
65 orgId: string,
66 exportId: string,
67 opts: { rowCap?: number } = {}
68): Promise<Record<string, unknown>> {
69 const t0 = Date.now();
70 const rowCap = opts.rowCap ?? ASYNC_ROW_CAP;
71 const key = `exports/${orgId}/${exportId}.csv`;
72 const mark = (patch: Record<string, unknown>) =>
73 prisma.auditExport.update({ where: { id: exportId }, data: { ...patch } as never });
74
75 try {
76 const row = await prisma.auditExport.findUnique({ where: { id: exportId } });
77 if (!row || row.organizationId !== orgId) {
78 throw new Error("export row not found (or cross-org)");
79 }
80 const filters = parseStoredFilters(row.filters);
81 const wsRows = await prisma.workspace.findMany({
82 where: { organizationId: orgId },
83 select: { id: true }
84 });
85 const wsIds = wsRows.map((w) => w.id);
86 const baseWhere = auditScopedWhere(orgId, wsIds, filters) as Prisma.AuditLogWhereInput;
87
88 const userNameCache = new Map<string, string>();
89 const wsNameCache = new Map<string, string>();
90 const lines: string[] = [AUDIT_CSV_HEADER];
91 let rows = 0;
92 let truncated = false;
93
94
95
96
97
98 let cursor: Prisma.AuditLogWhereInput = {};
99 for (;;) {
100 const batch = await prisma.auditLog.findMany({
101 where: { AND: [baseWhere, cursor] },
102 orderBy: [{ createdAt: "desc" }, { id: "desc" }],
103 take: BATCH
104 });
105 if (!batch.length) break;
106
107
108 const batchUsers = [...new Set(batch.filter((r) => r.actorType === "USER").map((r) => r.actorId))];
109 const batchWs = [...new Set(batch.map((r) => r.workspaceId).filter((w): w is string => Boolean(w)))];
110 if (batchUsers.length) {
111 const users = await prisma.user.findMany({
112 where: { id: { in: batchUsers } },
113 select: { id: true, name: true }
114 });
115 for (const u of users) userNameCache.set(u.id, u.name);
116 }
117 if (batchWs.length) {
118 const wss = await prisma.workspace.findMany({
119 where: { id: { in: batchWs } },
120 select: { id: true, name: true }
121 });
122 for (const w of wss) wsNameCache.set(w.id, w.name);
123 }
124
125 for (const r of batch) {
126 if (rows >= rowCap) {
127 truncated = true;
128 break;
129 }
130 rows++;
131 const cr: AuditCsvRow = {
132 id: r.id,
133 createdAt: r.createdAt,
134 actorType: r.actorType,
135 actorId: r.actorId,
136 workspaceId: r.workspaceId,
137 resourceType: r.resourceType,
138 resourceId: r.resourceId,
139 action: r.action,
140 ipAddress: r.ipAddress,
141 userAgent: r.userAgent,
142 metadata: r.metadata
143 };
144 lines.push(auditCsvLine(cr, cr.actorType === "USER" ? (userNameCache.get(cr.actorId) ?? "-") : "-", wsNameCache));
145 }
146 if (truncated) break;
147 if (batch.length < BATCH) break;
148 const last = batch[batch.length - 1]!;
149 cursor = {
150 OR: [
151 { createdAt: { lt: last.createdAt } },
152 { createdAt: last.createdAt, id: { lt: last.id } }
153 ]
154 };
155 }
156
157 const csv = lines.join("\n") + "\n";
158 const env = loadEnv();
159 const s3 = new S3Client({
160 endpoint: env.S3_ENDPOINT.replace(/\/+$/, ""),
161 region: env.S3_REGION,
162 bucket: env.S3_BUCKET_FILES || env.S3_BUCKET,
163 accessKey: env.S3_ACCESS_KEY,
164 secretKey: env.S3_SECRET_KEY
165 });
166 await s3.putObject(key, Buffer.from(csv, "utf8"), "text/csv; charset=utf-8");
167
168 await mark({ status: "ready", rows, truncated, completedAt: new Date(), error: null });
169 await prisma.auditLog.create({
170 data: {
171 actorType: "SYSTEM",
172 actorId: "worker:maintenance",
173 workspaceId: null,
174 resourceType: "organization",
175 resourceId: orgId,
176 action: "audit.export.completed",
177 metadata: asJson({ orgId, exportId, rows, truncated, key, durationMs: Date.now() - t0 })
178 }
179 });
180 log.info({ orgId, exportId, rows, truncated, durationMs: Date.now() - t0 }, "audit export ready");
181 return { exportId, status: "ready", rows, truncated, key };
182 } catch (err) {
183 const msg = (err as Error).message;
184 log.error({ orgId, exportId, err: msg }, "audit export failed");
185 await mark({ status: "failed", error: msg, completedAt: new Date() }).catch(() => undefined);
186 await prisma.auditLog
187 .create({
188 data: {
189 actorType: "SYSTEM",
190 actorId: "worker:maintenance",
191 workspaceId: null,
192 resourceType: "organization",
193 resourceId: orgId,
194 action: "audit.export.failed",
195 metadata: asJson({ orgId, exportId, error: msg })
196 }
197 })
198 .catch(() => undefined);
199 throw err;
200 }
201 void redis;
202}
203
