finalize.ts · Source

finalize.tsapps/worker/src/processors/files/steps/finalize.tsraw ← steps8,510 B← Back to the app
1/**
2 * apps/worker — files pipeline: finalize step (TASK-0102 §6 + §3).
3 *
4 * Persists the terminal state and publishes the event contract:
5 *
6 *   READY         — status READY + final metadata (sha256/dims/media/pdf +
7 *                   `pipeline`), `thumbnailKey` (480 variant), `scanReport`;
8 *                   `FileVersion` upsert (idempotent); `file.uploaded`
9 *                   emitted EXACTLY ONCE (only when this run actually
10 *                   transitioned the row to READY — the at-least-once /
11 *                   re-run dedupe); `file.finalized` (the TASK-0096 search
12 *                   re-upsert seam); audit `file.processed`.
13 *
14 *   VIRUS_DETECTED — status + `scanReport {engine:"clamav",version,infected,
15 *                   signature}`; `file.virus_detected` event + workspace-admin
16 *                   notification; audit. NO thumbnails, NO `file.uploaded`,
17 *                   NO search index update (the object stays stored for
18 *                   forensics; TASK-0103/0104 enforce download 403).
19 *
20 *   FAILED        — status + `metadata.pipeline.error` (+ code); audit
21 *                   `file.process_failed`.
22 */
23import { asJson } from "../../../lib/prisma.js";
24import { audit, notifyWorkspaceAdmins } from "../../../lib/notify.js";
25import { getLogger } from "../../../logger.js";
26import type { FilePipelineContext, PipelineMeta, PipelineOutcome } from "../types.js";
27import type { File, FileStatus } from "@prisma/client";
28 
29const log = getLogger("files.finalize");
30const STREAM = "factory:events";
31 
32export type FileRowLike = Pick<
33  File,
34  | "id"
35  | "workspaceId"
36  | "name"
37  | "key"
38  | "size"
39  | "mimeType"
40  | "status"
41  | "metadata"
42  | "thumbnailKey"
43  | "uploadedBy"
44  | "blockId"
45  | "pageId"
46>;
47 
48export interface FinalizeInput {
49  outcome: PipelineOutcome;
50  /** The row's status BEFORE this run's transition (read by the processor). */
51  prevStatus: FileStatus;
52  file: FileRowLike;
53  /** The authoritative (re-sniffed) mime for READY. */
54  finalMime: string;
55  /** The actual downloaded byte size. */
56  size: number;
57}
58 
59export async function finalizeStep(
60  ctx: FilePipelineContext,
61  input: FinalizeInput
62): Promise<void> {
63  const { outcome, prevStatus, file, finalMime, size } = input;
64  const { job } = ctx;
65  const meta = (file.metadata as Record<string, unknown> | null) ?? {};
66  const scannedAt = new Date().toISOString();
67  const steps = [...ctx.stepTimings.entries()].map(([name, ms]) => ({ name, ms }));
68 
69  if (outcome.kind === "virus") {
70    await ctx.prisma.file.update({
71      where: { id: job.fileId },
72      data: {
73        status: "VIRUS_DETECTED",
74        scanReport: asJson(outcome.scan as unknown as Record<string, unknown>),
75        metadata: asJson({ ...meta, pipeline: { scannedAt, error: "virus_detected" } as PipelineMeta })
76      }
77    });
78    await publish(ctx, "file.virus_detected", {
79      workspaceId: file.workspaceId,
80      fileId: job.fileId,
81      name: file.name,
82      signature: outcome.scan.signature,
83      key: file.key
84    });
85    await notifyWorkspaceAdmins(
86      ctx.prisma,
87      ctx.redis,
88      ctx.env.SMTP_HOST,
89      file.workspaceId,
90      "Malicious file blocked",
91      `Uploaded file "${file.name}" was flagged by ClamAV as ${outcome.scan.signature ?? "malicious"} and has been blocked.`,
92      { type: "file", id: job.fileId },
93      "AUTOMATION"
94    );
95    await audit(ctx.prisma, {
96      workspaceId: file.workspaceId,
97      resourceType: "file",
98      resourceId: job.fileId,
99      action: "file.virus_detected",
100      metadata: { signature: outcome.scan.signature, key: file.key, size }
101    });
102    log.error({ fileId: job.fileId, prevStatus }, "file finalized VIRUS_DETECTED");
103    return;
104  }
105 
106  if (outcome.kind === "fatal") {
107    const pipeline: Record<string, unknown> = {
108      scannedAt,
109      steps,
110      error: outcome.error,
111      ...(ctx.warnings.length ? { warnings: [...ctx.warnings] } : {})
112    };
113    await ctx.prisma.file.update({
114      where: { id: job.fileId },
115      data: {
116        status: "FAILED",
117        metadata: asJson({ ...meta, pipeline })
118      }
119    });
120    await audit(ctx.prisma, {
121      workspaceId: file.workspaceId,
122      resourceType: "file",
123      resourceId: job.fileId,
124      action: "file.process_failed",
125      metadata: { code: outcome.code, error: outcome.error, key: file.key }
126    });
127    log.error({ fileId: job.fileId, prevStatus, code: outcome.code }, "file finalized FAILED");
128    return;
129  }
130 
131  // ── READY ──────────────────────────────────────────────────────────────
132  const { extracted, thumbs, scan } = outcome;
133  const pipeline: PipelineMeta = {
134    scannedAt,
135    sha256: ctx.sha256,
136    durationMs: 0, // filled by the processor (total wall time)
137    steps,
138    ...(ctx.warnings.length ? { warnings: [...ctx.warnings] } : {})
139  };
140 
141  const metadata: Record<string, unknown> = { ...meta, sha256: ctx.sha256, pipeline };
142  if (extracted.dimensions) metadata.dimensions = extracted.dimensions;
143  if (extracted.media) metadata.media = extracted.media;
144  if (extracted.pdf) metadata.pdf = extracted.pdf;
145 
146  // `thumbnailUrl` (Requirement 5): a SHORT-LIVED signed GET of the 480
147  // artifact (15 min). Minting it here (not just `thumbnailKey`) is what makes
148  // the acceptance criterion "thumbnailUrl returns 200" true immediately after
149  // READY; the API re-mints on read (TASK-0103) so it never sits long-lived.
150  // Best-effort: a signing failure must not fail the run (the artifact + key
151  // exist and are re-mintable).
152  let thumbnailUrl: string | null = null;
153  if (thumbs.primaryKey) {
154    try {
155      thumbnailUrl = await ctx.s3Thumbs.presignedGet(thumbs.primaryKey, 15 * 60);
156    } catch (err) {
157      log.warn(
158        { fileId: job.fileId, err: (err as Error).message },
159        "thumbnail presign failed (thumbnailKey still set; re-mint on read)"
160      );
161    }
162  }
163 
164  await ctx.prisma.file.update({
165    where: { id: job.fileId },
166    data: {
167      status: "READY",
168      mimeType: finalMime,
169      size,
170      metadata: asJson(metadata),
171      scanReport: asJson(scan as unknown as Record<string, unknown>),
172      thumbnailKey: thumbs.primaryKey,
173      thumbnailUrl
174    }
175  });
176 
177  // FileVersion: TASK-0101 leaves the row in SCANNING without a version row;
178  // the worker owns the upsert (idempotent on [fileId, version]).
179  await ctx.prisma.fileVersion.upsert({
180    where: { fileId_version: { fileId: job.fileId, version: job.version } },
181    update: { key: file.key, size },
182    create: { fileId: job.fileId, version: job.version, key: file.key, size, createdBy: file.uploadedBy }
183  });
184 
185  // `file.uploaded` EXACTLY ONCE: only when THIS run made the SCANNING→READY
186  // (or UPLOADING→READY) transition. A re-run of an already-READY file
187  // returns before reaching here, so this guard is the at-least-once dedupe.
188  const isFreshReady = prevStatus === "SCANNING" || prevStatus === "UPLOADING";
189  if (isFreshReady) {
190    await publish(ctx, "file.uploaded", {
191      workspaceId: file.workspaceId,
192      fileId: job.fileId,
193      name: file.name,
194      mimeType: finalMime,
195      size,
196      ...(file.blockId ? { blockId: file.blockId } : {}),
197      ...(file.pageId ? { pageId: file.pageId } : {})
198    });
199  }
200  // TASK-0096 seam: the search re-upsert of the file doc with final metadata.
201  await publish(ctx, "file.finalized", {
202    workspaceId: file.workspaceId,
203    fileId: job.fileId,
204    mimeType: finalMime,
205    size
206  });
207 
208  await audit(ctx.prisma, {
209    workspaceId: file.workspaceId,
210    resourceType: "file",
211    resourceId: job.fileId,
212    action: "file.processed",
213    metadata: { key: file.key, sha256: ctx.sha256, mime: finalMime, size, steps }
214  });
215  log.info(
216    { fileId: job.fileId, prevStatus, mime: finalMime, size, thumbs: thumbs.keys.length, fresh: isFreshReady },
217    "file finalized READY"
218  );
219}
220 
221/** Best-effort stream publish (a lost event is repaired by the reconcile). */
222async function publish(
223  ctx: FilePipelineContext,
224  event: string,
225  payload: Record<string, unknown>
226): Promise<void> {
227  try {
228    await ctx.redis.xadd(
229      STREAM,
230      "MAXLEN",
231      "~",
232      "10000",
233      "*",
234      "event",
235      event,
236      "payload",
237      JSON.stringify(payload),
238      "ts",
239      new Date().toISOString()
240    );
241  } catch (err) {
242    log.warn({ err: (err as Error).message, event }, "event publish failed (reconcile backstops)");
243  }
244}
245