uploads.service.ts

uploads.service.tsapps/api/src/modules/files/uploads.service.tsraw ← files22,075 B← Back to the app
1import { Inject, Injectable } from "@nestjs/common";
2import { randomBytes } from "node:crypto";
3import type { Request } from "express";
4// Root-barrel import (NOT `@factory/storage/…` subpaths): the API compiles
5// with `moduleResolution: node10`, which resolves a package's `main`/`types`
6// but not its `exports` subpaths. The barrel re-exports every submodule.
7import {
8  buildFileKey,
9  isAllowedMime,
10  parseFileKey,
11  presignPut,
12  sanitizeFilename,
13  sniffVerdict,
14} from "@factory/storage";
15import { LIMITS } from "@factory/shared";
16import { jobTraceMetadata } from "../../common/trace-metadata.js";
17import { PrismaService } from "../../common/prisma.service.js";
18import { AuditService } from "../../common/audit.service.js";
19import { EventBus } from "../../common/event-bus.js";
20import { RedisService } from "../../common/redis.service.js";
21import { ApiError } from "../../common/errors.js";
22import { logger } from "../../common/logger.js";
23import { assertCan } from "../../lib/authorize.js";
24import { loadEnv } from "../../config/env.js";
25import { filesBucketConfig, makeFilesClient, signExpirySeconds } from "./s3.js";
26import {
27  PLAN_USAGE_LIMITS,
28  adjustWorkspaceStorageBytes,
29  authoritativeStoredBytes,
30  effectiveStorageQuota,
31} from "./quota.js";
32import { toFileRecord } from "./record.js";
33import { effectiveMaxFileBytes, workspaceOrgPolicy } from "../org/org-policy.service.js";
34 
35// Re-exported so existing importers (files-limits.test.ts) keep resolving the
36// plan-limit helpers from the module that historically owned them; the
37// canonical home is now modules/files/quota.ts (TASK-0104 §4 — shared by the
38// workspace usage endpoint too).
39export { PLAN_USAGE_LIMITS, effectiveStorageQuota } from "./quota.js";
40import type {
41  CancelUploadInput,
42  CompleteUploadInput,
43  CreateUploadInput,
44} from "./dto.js";
45 
46/**
47 * TASK-0101 — upload sessions (the API half of the two-phase upload).
48 *
49 *   POST   /files/uploads               → 201 { upload }
50 *   PUT    /files/uploads/:id/complete  → HEAD + 8 KiB sniff → SCANNING + job
51 *   POST   /files/uploads/:id/cancel    → delete partial object + row
52 *
53 * State machine (enforced here; the worker owns SCANNING → READY|…):
54 *   UPLOADING → SCANNING (complete) → (worker) READY | VIRUS_DETECTED | FAILED
55 *   UPLOADING → FAILED / deleted (cancel)
56 *
57 * Security model (spec §Security): the raw object key is NEVER returned to
58 * the client; only a query-string signed PUT URL (bound to the key + the
59 * declared content-type) leaves the server. The claimed mimeType is checked
60 * against the closed allow-list at session creation; the BYTES are checked
61 * against magic-byte sniffing at completion (sniffing is authoritative —
62 * mismatch = FAILED `MIME_MISMATCH` + object deleted, never "trust and warn").
63 */
64 
65/**
66 * Effective per-file size limit = min(usageLimits.maxFileSize, plan default,
67 * hard ceiling) — usageLimits is an admin OVERRIDE that may only tighten a
68 * plan's allowance, never exceed it. (The plan ceiling + the storage-quota
69 * min-rule now live in `./quota.ts`; re-exported above.)
70 */
71export function effectiveMaxFileSize(
72  billingPlan: string,
73  usageLimits: { maxFileSize?: number } | null,
74): number {
75  const plan = PLAN_USAGE_LIMITS[billingPlan] ?? PLAN_USAGE_LIMITS.free!;
76  const override = usageLimits?.maxFileSize;
77  return Math.min(
78    plan.maxFileSize,
79    typeof override === "number" && override > 0 ? override : Infinity,
80    LIMITS.MAX_FILE_SIZE,
81  );
82}
83 
84@Injectable()
85export class UploadsService {
86  constructor(
87    @Inject(PrismaService) private readonly prisma: PrismaService,
88    @Inject(AuditService) private readonly audit: AuditService,
89    @Inject(EventBus) private readonly events: EventBus,
90    @Inject(RedisService) private readonly redis: RedisService,
91  ) {}
92 
93  /**
94   * Open an upload session. Auth + EDIT on the anchor (page/block, or the
95   * workspace root when unattached), plan size limit (413 FILE_TOO_LARGE),
96   * storage quota (413 QUOTA_EXCEEDED), closed MIME allow-list
97   * (400 mime_not_allowed). Creates the File row (UPLOADING) and returns a
98   * query-string signed PUT URL (900 s, content-type bound).
99   */
100  async create(userId: string, input: CreateUploadInput, req: Request) {
101    const workspaceId = this.workspaceOf(req);
102    await assertCan(this.prisma, { workspaceId, userId }, "EDIT");
103 
104    const ws = await this.prisma.workspace.findUnique({ where: { id: workspaceId } });
105    if (!ws || ws.deletedAt) throw ApiError.notFound("Workspace not found.");
106 
107    // Resolve the anchor (block ⇒ its page) and require EDIT there — the
108    // file inherits the anchor's permission scope (spec req. 3). `null` (an
109    // explicit "not attached", CreateUploadDto) is normalized to `undefined`.
110    let pageId: string | undefined = input.pageId ?? undefined;
111    let blockId: string | undefined = input.blockId ?? undefined;
112    if (blockId) {
113      const block = await this.prisma.block.findUnique({ where: { id: blockId } });
114      if (!block || block.pageId !== pageId) {
115        throw ApiError.notFound("Anchor block not found in this workspace.");
116      }
117      pageId = block.pageId;
118    }
119    if (pageId) {
120      const page = await this.prisma.page.findUnique({ where: { id: pageId } });
121      if (!page || page.workspaceId !== workspaceId) {
122        throw ApiError.notFound("Anchor page not found in this workspace.");
123      }
124    }
125    // A block anchors to its page (authorization scopes by page — blocks
126    // have no scope of their own); `blockId` is kept for the File row + audit.
127    await assertCan(this.prisma, {
128      workspaceId,
129      userId,
130      resource: { pageId },
131    }, "EDIT");
132 
133    // ── plan limits ────────────────────────────────────────────────────
134    const limits = (ws.usageLimits ?? {}) as { maxFileSize?: number; storageQuota?: number };
135    const planMaxFileSize = effectiveMaxFileSize(ws.billingPlan, limits);
136    // TASK-0133 enforcement #3: an org `maxFileSizeMB` policy CAPS the plan
137    // allowance (org cap ∩ plan cap — the policy can only TIGHTEN, never
138    // raise). Authoritative (413) — the file picker's warning is UX only.
139    // (The policy read is FRESH, no cache — a security check.)
140    const { policy } = await workspaceOrgPolicy(this.prisma, workspaceId);
141    const maxFileSize = Math.min(
142      planMaxFileSize,
143      effectiveMaxFileBytes(policy, ws.billingPlan),
144    );
145    if (input.size > maxFileSize) {
146      throw new ApiError("FILE_TOO_LARGE", 413,
147        `File size ${input.size} exceeds the ${maxFileSize}-byte limit for this workspace.`,
148        [{ maxBytes: maxFileSize, plan: ws.billingPlan, orgPolicy: maxFileSize < planMaxFileSize }]);
149    }
150    const quota = effectiveStorageQuota(ws.billingPlan, limits);
151    // TASK-0104 §4/§Security: the counter is ADVISORY-FAST (O(1), never a
152    // live sum on the happy path). Only at the DENIAL boundary is the
153    // authoritative sum re-checked, and only a stale-LOW counter (drift >
154    // 1 GB — e.g. the one-time `Σ File.size` backfill vs the FileVersion
155    // ledger) is corrected to the sum before denying. A stale-HIGH counter
156    // (conservative) is repaired by the maintenance reconciler.
157    let used = Number(ws.usageStorageBytes ?? 0n);
158    if (input.size > quota - used) {
159      try {
160        const authoritative = Number(await authoritativeStoredBytes(this.prisma, workspaceId));
161        if (authoritative - used > 1024 ** 3) {
162          used = authoritative;
163          await this.prisma.workspace.update({
164            where: { id: workspaceId },
165            data: { usageStorageBytes: BigInt(authoritative) },
166          });
167        }
168      } catch (err) {
169        // The re-check is defensive; a blip keeps the counter decision
170        // (the conservative one — it can only over-deny, never under-deny).
171        logger.warn(`quota authoritative re-check failed for ws ${workspaceId}: ${(err as Error).message}`);
172      }
173    }
174    if (input.size > quota - used) {
175      throw new ApiError("QUOTA_EXCEEDED", 413,
176        `Uploading this file would exceed the workspace storage quota (${quota} bytes; ${used} used).`,
177        [{ maxBytes: quota, usedBytes: used, remainingBytes: quota - used }]);
178    }
179 
180    // ── closed MIME allow-list (claimed type) ──────────────────────────
181    if (!isAllowedMime(input.mimeType)) {
182      throw ApiError.validation("File type is not allowed.", [
183        { field: "mimeType", issue: "mime_not_allowed", received: input.mimeType },
184      ]);
185    }
186 
187    // ── File row + signed PUT URL ──────────────────────────────────────
188    // TASK-0104 §1: `replaceFileId` makes the session a NEW VERSION of an
189    // existing file (the UI "Replace file" affordance) instead of a fresh
190    // file. The existing File row is re-pointed at the new versioned key;
191    // the old object is RETAINED (a FileVersion row for it stays until purge
192    // or "delete old versions").
193    const replaceFileId: string | undefined = input.replaceFileId ?? undefined;
194    let target: {
195      id: string;
196      workspaceId: string;
197      name: string;
198      pageId: string | null;
199      blockId: string | null;
200      status: string;
201      deletedAt: Date | null;
202    } | null = null;
203    let nextVersion = 1;
204    if (replaceFileId) {
205      target = await this.prisma.file.findUnique({
206        where: { id: replaceFileId },
207        select: {
208          id: true,
209          workspaceId: true,
210          name: true,
211          pageId: true,
212          blockId: true,
213          status: true,
214          deletedAt: true,
215        },
216      });
217      if (!target || target.workspaceId !== workspaceId || target.deletedAt) {
218        throw ApiError.notFound("File to replace not found in this workspace.");
219      }
220      // A replace is an EDIT of the file (file + page scope, TASK-0104).
221      await assertCan(this.prisma, {
222        workspaceId,
223        userId,
224        resource: { fileId: target.id, pageId: target.pageId ?? undefined },
225      }, "EDIT");
226      // The file must be in a stable state before a new version is layered on
227      // (a still-scanning version's bytes would race the re-point).
228      if (target.status === "UPLOADING" || target.status === "SCANNING") {
229        throw ApiError.conflict("This file is still processing; wait for it to finish before replacing.", [
230          { reason: "replace_while_processing" },
231        ]);
232      }
233      const latest = await this.prisma.fileVersion.aggregate({
234        where: { fileId: target.id },
235        _max: { version: true },
236      });
237      nextVersion = (latest._max.version ?? 0) + 1;
238      // A replace re-anchors to the EXISTING file's page/block (the request's
239      // anchors are ignored — the file already has a home).
240      pageId = target.pageId ?? undefined;
241      blockId = target.blockId ?? undefined;
242    }
243 
244    const env = loadEnv();
245    const fileId = target?.id ?? `file_${randomBytes(12).toString("base64url")}`;
246    const key = buildFileKey({
247      workspaceId,
248      fileId,
249      version: nextVersion,
250      name: sanitizeFilename(input.name),
251    });
252    const now = new Date();
253    const expiresSec = signExpirySeconds(env);
254 
255    let file;
256    if (target) {
257      // Re-point the existing row at the new version (it re-enters the
258      // pipeline via complete → SCANNING; the worker upserts the FileVersion).
259      file = await this.prisma.file.update({
260        where: { id: target.id },
261        data: {
262          name: input.name,
263          key,
264          size: input.size,
265          mimeType: input.mimeType,
266          status: "UPLOADING",
267          metadata: {},
268          uploadedBy: userId,
269        },
270      });
271    } else {
272      file = await this.prisma.file.create({
273        data: {
274          id: fileId,
275          workspaceId,
276          pageId: pageId ?? null,
277          blockId: blockId ?? null,
278          name: input.name,
279          key,
280          size: input.size,
281          mimeType: input.mimeType,
282          status: "UPLOADING",
283          metadata: {},
284          uploadedBy: userId,
285        },
286      });
287    }
288 
289    const put = presignPut({
290      config: filesBucketConfig(env),
291      key,
292      contentType: input.mimeType,
293      expiresSec,
294      now,
295    });
296    const expiresAt = new Date(now.getTime() + expiresSec * 1000);
297 
298    await this.audit.audit(req, req.auth, {
299      action: "file.upload_session_created",
300      resourceType: "file",
301      resourceId: file.id,
302      workspaceId,
303      metadata: { name: input.name, size: input.size, mimeType: input.mimeType, pageId, blockId },
304    });
305 
306    return {
307      upload: {
308        file: toFileRecord(file, 1),
309        uploadUrl: put.url,
310        method: "PUT" as const,
311        headers: {
312          // The SIGNED header set the client must send on the PUT.
313          //
314          // content-type is the ONE signed client header: it is signed into
315          // the URL (the "wrong Content-Type → 403" guarantee), so the client
316          // must send it back EXACTLY. Do NOT also spread `put.headers` — it
317          // carries the same content-type in lower case and the client would
318          // send a DUPLICATE header (undici joins the values, breaking the
319          // signature → MinIO 403 SignatureDoesNotMatch).
320          //
321          // No Content-Length here, deliberately: this signer does NOT
322          // content-length-lock the presigned PUT (the spec allows the lock
323          // only "when the S3 SDK supports it"), and returning
324          // `Content-Length: <declared>` would make strict HTTP clients
325          // (undici) REFUSE to send a body whose size differs from the
326          // declared one — which is precisely the "PUT fewer bytes than
327          // declared" case that must reach `complete` and 422
328          // UPLOAD_INCOMPLETE. undici/fetch set Content-Length from the
329          // actual body automatically.
330          "Content-Type": input.mimeType,
331        },
332        expiresAt: expiresAt.toISOString(),
333      },
334    };
335  }
336 
337  /**
338   * Complete a session: HEAD the object (size must match the declaration —
339   * otherwise 422 UPLOAD_INCOMPLETE), read the first 8 KiB and run
340   * `sniffMagicBytes`; a mismatch with the CLAIMED type marks the file
341   * FAILED (`MIME_MISMATCH`) and deletes the object (the bytes never
342   * become downloadable). On success: status SCANNING, enqueue the
343   * `files` queue job `process-file { fileId, version }`, audit + publish.
344   */
345  async complete(uploadId: string, userId: string, _input: CompleteUploadInput, req: Request) {
346    const file = await this.requireUploadable(uploadId, userId, "EDIT");
347    const s3 = makeFilesClient();
348 
349    let head: { size: number };
350    try {
351      head = await s3.headObject(file.key);
352    } catch (err) {
353      const status = (err as { status?: number })?.status;
354      if (status === 404) {
355        // Object never landed (signed window closed, or the object was
356        // already reaped): the session can no longer complete.
357        throw new ApiError("UPLOAD_EXPIRED", 410,
358          "The upload object is not present (the signed URL expired or the upload never completed).",
359          [{ reason: "upload_expired" }]);
360      }
361      throw err;
362    }
363 
364    if (head.size !== file.size) {
365      throw new ApiError("UPLOAD_INCOMPLETE", 422,
366        `Uploaded object size ${head.size} does not match the declared size ${file.size}.`,
367        [{ received: head.size, expected: file.size }]);
368    }
369 
370    // Authoritative magic-byte check on the FIRST 8 KiB (the verdict rules —
371    // signature-bearing claims must match, signature-less claims pass — live
372    // in the storage package so the worker and the API can't drift).
373    // Storage blip on the sniff read propagates as-is (retryable at transport).
374    const { prefix } = await s3.readPrefix(file.key, 8 * 1024);
375    const verdict = sniffVerdict(prefix, file.mimeType);
376    if (verdict === "mismatch") {
377      try {
378        await s3.deleteObject(file.key);
379      } catch {
380        // The row is FAILED and never signed; a stray object is reaped by
381        // the orphan sweeper (TASK-0104).
382      }
383      await this.prisma.file.update({
384        where: { id: file.id },
385        data: {
386          status: "FAILED",
387          metadata: {
388            ...(file.metadata as Record<string, unknown>),
389            pipeline: { error: "mime_mismatch" },
390          },
391          scanReport: { engine: "magic-bytes", result: "mismatch", claimed: file.mimeType },
392        },
393      });
394      await this.audit.audit(req, req.auth, {
395        action: "file.mime_mismatch",
396        resourceType: "file",
397        resourceId: file.id,
398        workspaceId: file.workspaceId,
399        metadata: { name: file.name, claimed: file.mimeType },
400      });
401      throw new ApiError("MIME_MISMATCH", 415,
402        "The uploaded bytes do not match the declared MIME type; the object was deleted.",
403        [{ claimed: file.mimeType, issue: "mime_mismatch" }]);
404    }
405 
406    // TASK-0104: the version is encoded in the object key (buildFileKey puts
407    // it at the third path segment). For a fresh upload it is 1; for a
408    // `replaceFileId` session it is n+1. The worker upserts the FileVersion
409    // row at finalize using this version.
410    let version = 1;
411    try {
412      version = parseFileKey(file.key).version;
413    } catch {
414      version = 1; // legacy/malformed key — fall back to the original behavior
415    }
416 
417    // TASK-0104 §4: the bytes land in the bucket at `complete` (they are
418    // durable from here — the worker finalizes metadata/thumbs, but the
419    // object already exists). Count them in the SAME transaction as the
420    // SCANNING transition (no drift window): +head.size. A later prune/purge
421    // decrements (files.service); restore changes nothing (bytes retained).
422    const updated = await this.prisma.$transaction(async (tx) => {
423      const row = await tx.file.update({
424        where: { id: file.id },
425        data: { status: "SCANNING", size: head.size },
426      });
427      await adjustWorkspaceStorageBytes(tx, file.workspaceId, BigInt(head.size));
428      return row;
429    });
430 
431    // Enqueue the pipeline job (worker does scan/metadata/thumbnails).
432    await this.enqueueProcessFile(file.id, version, file.workspaceId);
433 
434    await this.audit.audit(req, req.auth, {
435      action: version > 1 ? "file.version_created" : "file.uploaded",
436      resourceType: "file",
437      resourceId: file.id,
438      workspaceId: file.workspaceId,
439      metadata: { name: file.name, size: head.size, status: updated.status, version },
440    });
441    await this.events.publish("file.uploaded", {
442      workspaceId: file.workspaceId,
443      fileId: file.id,
444      name: file.name,
445      size: head.size,
446      mimeType: file.mimeType,
447      blockId: file.blockId,
448      pageId: file.pageId,
449      version,
450    });
451 
452    return { id: file.id, status: updated.status, verifiedBy: "worker", version };
453  }
454 
455  /**
456   * Cancel a session: delete the partial object (if any) and the File row
457   * (hard — the session never became a file).
458   */
459  async cancel(uploadId: string, userId: string, _input: CancelUploadInput, req: Request) {
460    const file = await this.requireUploadable(uploadId, userId, "EDIT");
461    const s3 = makeFilesClient();
462    try {
463      await s3.deleteObject(file.key);
464    } catch {
465      // Object absent — nothing to delete.
466    }
467    await this.prisma.file.delete({ where: { id: file.id } });
468    await this.audit.audit(req, req.auth, {
469      action: "file.upload_cancelled",
470      resourceType: "file",
471      resourceId: file.id,
472      workspaceId: file.workspaceId,
473      metadata: { name: file.name, key: file.key },
474    });
475    return { deleted: true };
476  }
477 
478  // ── internals ────────────────────────────────────────────────────────
479 
480  private workspaceOf(req: Request): string {
481    const ws = req.auth?.workspaceId;
482    if (!ws) {
483      throw ApiError.validation("Missing workspace context (x-factory-workspace).", [
484        { field: "workspace", issue: "workspace_required" },
485      ]);
486    }
487    return ws;
488  }
489 
490  /** The upload session row, or 404; completed sessions cannot be re-driven. */
491  private async requireUploadable(fileId: string, userId: string, need: "VIEW" | "EDIT") {
492    const file = await this.prisma.file.findUnique({ where: { id: fileId } });
493    if (!file || file.deletedAt) throw ApiError.notFound("Upload session not found.");
494    if (file.status === "SCANNING" || file.status === "READY" || file.status === "VIRUS_DETECTED") {
495      throw ApiError.conflict("This upload session has already been completed.", [
496        { reason: "upload_already_completed" },
497      ]);
498    }
499    await assertCan(this.prisma, {
500      workspaceId: file.workspaceId,
501      userId,
502      resource: { pageId: file.pageId ?? undefined },
503    }, need);
504    return file;
505  }
506 
507  /** Enqueue `process-file` on the `files` BullMQ queue. Degrades on a
508   *  Redis blip (logged) — the maintenance sweeper re-drives stranded
509   *  SCANNING rows, so the completion itself stays available. */
510  private async enqueueProcessFile(fileId: string, version: number, workspaceId: string): Promise<void> {
511    try {
512      const { Queue } = await import("bullmq");
513      const q = new Queue("files", {
514        connection: this.redis.getRedis().options as never,
515        defaultJobOptions: {
516          attempts: 3,
517          backoff: { type: "exponential", delay: 2_000 },
518          removeOnComplete: 500,
519          removeOnFail: 200,
520        },
521      });
522      // TASK-0135: continue the enqueueing request's trace into the job.
523      await q.add("process-file", { fileId, version, workspaceId }, { ...jobTraceMetadata() });
524      await q.close();
525    } catch (err) {
526      logger.warn(`files queue enqueue failed for ${fileId}: ${(err as Error).message}`);
527    }
528  }
529}
530