feat: chunk uploads (#76)

* add first concept

* finished first concept

* allow 3 uploads at same time

* retry if chunk failed

* updated clean temporary files job

* fix throttling for chunk uploads

* update tests

* remove multer

* migrate from `MAX_FILE_SIZE` to `MAX_SHARE_SIZE`

* improve error handling if file failed to upload

* fix promise limit

* improve file progress

backend/src/app.module.ts

@@ -1,19 +1,17 @@
-import { HttpException, HttpStatus, Module } from "@nestjs/common";
+import { Module } from "@nestjs/common";
 
 import { ScheduleModule } from "@nestjs/schedule";
 import { AuthModule } from "./auth/auth.module";
 
-import { MulterModule } from "@nestjs/platform-express";
-import { ThrottlerModule } from "@nestjs/throttler";
-import { Request } from "express";
+import { APP_GUARD } from "@nestjs/core";
+import { ThrottlerGuard, ThrottlerModule } from "@nestjs/throttler";
 import { ConfigModule } from "./config/config.module";
-import { ConfigService } from "./config/config.service";
 import { EmailModule } from "./email/email.module";
 import { FileModule } from "./file/file.module";
+import { JobsModule } from "./jobs/jobs.module";
 import { PrismaModule } from "./prisma/prisma.module";
 import { ShareModule } from "./share/share.module";
 import { UserModule } from "./user/user.module";
-import { JobsModule } from "./jobs/jobs.module";
 
 @Module({
   imports: [
@@ -25,29 +23,17 @@ import { JobsModule } from "./jobs/jobs.module";
     ConfigModule,
     JobsModule,
     UserModule,
-    MulterModule.registerAsync({
-      useFactory: (config: ConfigService) => ({
-        fileFilter: (req: Request, file, cb) => {
-          const MAX_FILE_SIZE = config.get("MAX_FILE_SIZE");
-          const requestFileSize = parseInt(req.headers["content-length"]);
-          const isValidFileSize = requestFileSize <= MAX_FILE_SIZE;
-          cb(
-            !isValidFileSize &&
-              new HttpException(
-                `File must be smaller than ${MAX_FILE_SIZE} bytes`,
-                HttpStatus.PAYLOAD_TOO_LARGE
-              ),
-            isValidFileSize
-          );
-        },
-      }),
-      inject: [ConfigService],
-    }),
     ThrottlerModule.forRoot({
       ttl: 60,
       limit: 100,
     }),
     ScheduleModule.forRoot(),
   ],
+  providers: [
+    {
+      provide: APP_GUARD,
+      useClass: ThrottlerGuard,
+    },
+  ],
 })
 export class AppModule {}

backend/src/file/file.controller.ts

@@ -1,20 +1,19 @@
 import {
+  Body,
   Controller,
   Get,
   Param,
   Post,
+  Query,
   Res,
   StreamableFile,
-  UploadedFile,
   UseGuards,
-  UseInterceptors,
 } from "@nestjs/common";
-import { FileInterceptor } from "@nestjs/platform-express";
+import { SkipThrottle } from "@nestjs/throttler";
 import * as contentDisposition from "content-disposition";
 import { Response } from "express";
 import { JwtGuard } from "src/auth/guard/jwt.guard";
 import { FileDownloadGuard } from "src/file/guard/fileDownload.guard";
-import { ShareDTO } from "src/share/dto/share.dto";
 import { ShareOwnerGuard } from "src/share/guard/shareOwner.guard";
 import { ShareSecurityGuard } from "src/share/guard/shareSecurity.guard";
 import { FileService } from "./file.service";
@@ -24,22 +23,24 @@ export class FileController {
   constructor(private fileService: FileService) {}
 
   @Post()
+  @SkipThrottle()
   @UseGuards(JwtGuard, ShareOwnerGuard)
-  @UseInterceptors(
-    FileInterceptor("file", {
-      dest: "./data/uploads/_temp/",
-    })
-  )
   async create(
-    @UploadedFile()
-    file: Express.Multer.File,
+    @Query() query: any,
+
+    @Body() body: string,
     @Param("shareId") shareId: string
   ) {
-    // Fixes file names with special characters
-    file.originalname = Buffer.from(file.originalname, "latin1").toString(
-      "utf8"
+    const { id, name, chunkIndex, totalChunks } = query;
+
+    const data = body.toString().split(",")[1];
+
+    return await this.fileService.create(
+      data,
+      { index: parseInt(chunkIndex), total: parseInt(totalChunks) },
+      { id, name },
+      shareId
     );
-    return new ShareDTO().from(await this.fileService.create(file, shareId));
   }
 
   @Get(":fileId/download")

backend/src/file/file.module.ts

@@ -3,12 +3,11 @@ import { JwtModule } from "@nestjs/jwt";
 import { ShareModule } from "src/share/share.module";
 import { FileController } from "./file.controller";
 import { FileService } from "./file.service";
-import { FileValidationPipe } from "./pipe/fileValidation.pipe";
 
 @Module({
   imports: [JwtModule.register({}), ShareModule],
   controllers: [FileController],
-  providers: [FileService, FileValidationPipe],
+  providers: [FileService],
   exports: [FileService],
 })
 export class FileModule {}

backend/src/file/file.service.ts

@@ -1,10 +1,12 @@
 import {
   BadRequestException,
+  HttpException,
+  HttpStatus,
   Injectable,
   NotFoundException,
 } from "@nestjs/common";
 import { JwtService } from "@nestjs/jwt";
-import { randomUUID } from "crypto";
+import * as crypto from "crypto";
 import * as fs from "fs";
 import * as mime from "mime-types";
 import { ConfigService } from "src/config/config.service";
@@ -18,32 +20,85 @@ export class FileService {
     private config: ConfigService
   ) {}
 
-  async create(file: Express.Multer.File, shareId: string) {
+  async create(
+    data: string,
+    chunk: { index: number; total: number },
+    file: { id?: string; name: string },
+    shareId: string
+  ) {
+    if (!file.id) file.id = crypto.randomUUID();
+
     const share = await this.prisma.share.findUnique({
       where: { id: shareId },
+      include: { files: true },
     });
 
     if (share.uploadLocked)
       throw new BadRequestException("Share is already completed");
 
-    const fileId = randomUUID();
+    let diskFileSize: number;
+    try {
+      diskFileSize = fs.statSync(
+        `./data/uploads/shares/${shareId}/${file.id}.tmp-chunk`
+      ).size;
+    } catch {
+      diskFileSize = 0;
+    }
 
-    await fs.promises.mkdir(`./data/uploads/shares/${shareId}`, {
-      recursive: true,
-    });
-    fs.promises.rename(
-      `./data/uploads/_temp/${file.filename}`,
-      `./data/uploads/shares/${shareId}/${fileId}`
+    // If the sent chunk index and the expected chunk index doesn't match throw an error
+    const chunkSize = 10 * 1024 * 1024; // 10MB
+    const expectedChunkIndex = Math.ceil(diskFileSize / chunkSize);
+
+    if (expectedChunkIndex != chunk.index)
+      throw new BadRequestException({
+        message: "Unexpected chunk received",
+        error: "unexpected_chunk_index",
+        expectedChunkIndex,
+      });
+
+    const buffer = Buffer.from(data, "base64");
+
+    // Check if share size limit is exceeded
+    const fileSizeSum = share.files.reduce(
+      (n, { size }) => n + parseInt(size),
+      0
     );
 
-    return await this.prisma.file.create({
-      data: {
-        id: fileId,
-        name: file.originalname,
-        size: file.size.toString(),
-        share: { connect: { id: shareId } },
-      },
-    });
+    if (
+      fileSizeSum + diskFileSize + buffer.byteLength >
+      this.config.get("MAX_SHARE_SIZE")
+    ) {
+      throw new HttpException(
+        "Max share size exceeded",
+        HttpStatus.PAYLOAD_TOO_LARGE
+      );
+    }
+
+    fs.appendFileSync(
+      `./data/uploads/shares/${shareId}/${file.id}.tmp-chunk`,
+      buffer
+    );
+
+    const isLastChunk = chunk.index == chunk.total - 1;
+    if (isLastChunk) {
+      fs.renameSync(
+        `./data/uploads/shares/${shareId}/${file.id}.tmp-chunk`,
+        `./data/uploads/shares/${shareId}/${file.id}`
+      );
+      const fileSize = fs.statSync(
+        `./data/uploads/shares/${shareId}/${file.id}`
+      ).size;
+      await this.prisma.file.create({
+        data: {
+          id: file.id,
+          name: file.name,
+          size: fileSize.toString(),
+          share: { connect: { id: shareId } },
+        },
+      });
+    }
+
+    return file;
   }
 
   async get(shareId: string, fileId: string) {

backend/src/file/pipe/fileValidation.pipe.ts

@@ -1,17 +0,0 @@
-import {
-  ArgumentMetadata,
-  BadRequestException,
-  Injectable,
-  PipeTransform,
-} from "@nestjs/common";
-import { ConfigService } from "src/config/config.service";
-
-@Injectable()
-export class FileValidationPipe implements PipeTransform {
-  constructor(private config: ConfigService) {}
-  async transform(value: any, metadata: ArgumentMetadata) {
-    if (value.size > this.config.get("MAX_FILE_SIZE"))
-      throw new BadRequestException("File is ");
-    return value;
-  }
-}

backend/src/jobs/jobs.service.ts

@@ -38,18 +38,34 @@ export class JobsService {
 
   @Cron("0 0 * * *")
   deleteTemporaryFiles() {
-    const files = fs.readdirSync("./data/uploads/_temp");
+    let filesDeleted = 0;
 
-    for (const file of files) {
-      const stats = fs.statSync(`./data/uploads/_temp/${file}`);
-      const isOlderThanOneDay = moment(stats.mtime)
-        .add(1, "day")
-        .isBefore(moment());
+    const shareDirectories = fs
+      .readdirSync("./data/uploads/shares", { withFileTypes: true })
+      .filter((dirent) => dirent.isDirectory())
+      .map((dirent) => dirent.name);
 
-      if (isOlderThanOneDay) fs.rmSync(`./data/uploads/_temp/${file}`);
+    for (const shareDirectory of shareDirectories) {
+      const temporaryFiles = fs
+        .readdirSync(`./data/uploads/shares/${shareDirectory}`)
+        .filter((file) => file.endsWith(".tmp-chunk"));
+
+      for (const file of temporaryFiles) {
+        const stats = fs.statSync(
+          `./data/uploads/shares/${shareDirectory}/${file}`
+        );
+        const isOlderThanOneDay = moment(stats.mtime)
+          .add(1, "day")
+          .isBefore(moment());
+
+        if (isOlderThanOneDay) {
+          fs.rmSync(`./data/uploads/shares/${shareDirectory}/${file}`);
+          filesDeleted++;
+        }
+      }
     }
 
-    console.log(`job: deleted ${files.length} temporary files`);
+    console.log(`job: deleted ${filesDeleted} temporary files`);
   }
 
   @Cron("0 * * * *")

backend/src/main.ts

@@ -1,6 +1,7 @@
 import { ClassSerializerInterceptor, ValidationPipe } from "@nestjs/common";
 import { NestFactory, Reflector } from "@nestjs/core";
 import { NestExpressApplication } from "@nestjs/platform-express";
+import * as bodyParser from "body-parser";
 import * as cookieParser from "cookie-parser";
 import * as fs from "fs";
 import { AppModule } from "./app.module";
@@ -10,6 +11,7 @@ async function bootstrap() {
   app.useGlobalPipes(new ValidationPipe({ whitelist: true }));
   app.useGlobalInterceptors(new ClassSerializerInterceptor(app.get(Reflector)));
 
+  app.use(bodyParser.raw({type:'application/octet-stream', limit:'20mb'}));
   app.use(cookieParser());
   app.set("trust proxy", true);
 

backend/src/share/share.service.ts

@@ -56,6 +56,10 @@ export class ShareService {
       expirationDate = moment(0).toDate();
     }
 
+    fs.mkdirSync(`./data/uploads/shares/${share.id}`, {
+      recursive: true,
+    });
+
     return await this.prisma.share.create({
       data: {
         ...share,