feat: allow upload of large files (#1862)

* Allow upload of large files

* feat: createByteCountingStream utility function.

---------

Co-authored-by: gpapp <gergely.papp@itworks.hu>

apps/server/src/integrations/import/services/import.service.ts

@@ -10,7 +10,11 @@ import {
 } from '../../../collaboration/collaboration.util';
 import { InjectKysely } from 'nestjs-kysely';
 import { KyselyDB } from '@docmost/db/types/kysely.types';
-import { generateSlugId, sanitizeFileName } from '../../../common/helpers';
+import {
+  generateSlugId,
+  sanitizeFileName,
+  createByteCountingStream,
+} from '../../../common/helpers';
 import { generateJitteredKeyBetween } from 'fractional-indexing-jittered';
 import { TiptapTransformer } from '@hocuspocus/transformer';
 import * as Y from 'yjs';
@@ -173,15 +177,24 @@ export class ImportService {
     };
   }
 
-  async getNewPagePosition(spaceId: string): Promise<string> {
-    const lastPage = await this.db
+  async getNewPagePosition(
+    spaceId: string,
+    parentPageId?: string,
+  ): Promise<string> {
+    let query = this.db
       .selectFrom('pages')
       .select(['id', 'position'])
       .where('spaceId', '=', spaceId)
       .orderBy('position', (ob) => ob.collate('C').desc())
-      .limit(1)
-      .where('parentPageId', 'is', null)
-      .executeTakeFirst();
+      .limit(1);
+
+    if (parentPageId) {
+      query = query.where('parentPageId', '=', parentPageId);
+    } else {
+      query = query.where('parentPageId', 'is', null);
+    }
+
+    const lastPage = await query.executeTakeFirst();
 
     if (lastPage) {
       return generateJitteredKeyBetween(lastPage.position, null);
@@ -198,20 +211,21 @@ export class ImportService {
     workspaceId: string,
   ) {
     const file = await filePromise;
-    const fileBuffer = await file.toBuffer();
     const fileExtension = path.extname(file.filename).toLowerCase();
     const fileName = sanitizeFileName(
       path.basename(file.filename, fileExtension),
     );
-    const fileSize = fileBuffer.length;
-
     const fileNameWithExt = fileName + fileExtension;
 
     const fileTaskId = uuid7();
     const filePath = `${getFileTaskFolderPath(FileTaskType.Import, workspaceId)}/${fileTaskId}/${fileNameWithExt}`;
 
     // upload file
-    await this.storageService.upload(filePath, fileBuffer);
+    const { stream, getBytesRead } = createByteCountingStream(file.file);
+
+    await this.storageService.upload(filePath, stream);
+
+    const fileSize = getBytesRead();
 
     const fileTask = await this.db
       .insertInto('fileTasks')

apps/server/src/integrations/storage/drivers/local.driver.ts

@@ -20,9 +20,15 @@ export class LocalDriver implements StorageDriver {
     return join(this.config.storagePath, filePath);
   }
 
-  async upload(filePath: string, file: Buffer): Promise<void> {
+  async upload(filePath: string, file: Buffer | Readable): Promise<void> {
     try {
-      await fs.outputFile(this._fullPath(filePath), file);
+      const fullPath = this._fullPath(filePath);
+      if (file instanceof Buffer) {
+        await fs.outputFile(fullPath, file);
+      } else {
+        await fs.mkdir(dirname(fullPath), { recursive: true });
+        await pipeline(file, createWriteStream(fullPath));
+      }
     } catch (err) {
       throw new Error(`Failed to upload file: ${(err as Error).message}`);
     }
@@ -42,7 +48,7 @@ export class LocalDriver implements StorageDriver {
     try {
       const fromFullPath = this._fullPath(fromFilePath);
       const toFullPath = this._fullPath(toFilePath);
-      
+
       if (await this.exists(fromFilePath)) {
         await fs.copy(fromFullPath, toFullPath);
       }

apps/server/src/integrations/storage/drivers/s3.driver.ts

@@ -23,19 +23,21 @@ export class S3Driver implements StorageDriver {
     this.s3Client = new S3Client(config as any);
   }
 
-  async upload(filePath: string, file: Buffer): Promise<void> {
+  async upload(filePath: string, file: Buffer | Readable): Promise<void> {
     try {
       const contentType = getMimeType(filePath);
 
-      const command = new PutObjectCommand({
-        Bucket: this.config.bucket,
-        Key: filePath,
-        Body: file,
-        ContentType: contentType,
-        // ACL: "public-read",
+      const upload = new Upload({
+        client: this.s3Client,
+        params: {
+          Bucket: this.config.bucket,
+          Key: filePath,
+          Body: file,
+          ContentType: contentType,
+        },
       });
 
-      await this.s3Client.send(command);
+      await upload.done();
     } catch (err) {
       throw new Error(`Failed to upload file: ${(err as Error).message}`);
     }

apps/server/src/integrations/storage/interfaces/storage-driver.interface.ts

@@ -1,7 +1,7 @@
 import { Readable } from 'stream';
 
 export interface StorageDriver {
-  upload(filePath: string, file: Buffer): Promise<void>;
+  upload(filePath: string, file: Buffer | Readable): Promise<void>;
 
   uploadStream(filePath: string, file: Readable, options?: { recreateClient?: boolean }): Promise<void>;
 

apps/server/src/integrations/storage/storage.service.ts

@@ -8,9 +8,9 @@ export class StorageService {
   private readonly logger = new Logger(StorageService.name);
   constructor(
     @Inject(STORAGE_DRIVER_TOKEN) private storageDriver: StorageDriver,
-  ) {}
+  ) { }
 
-  async upload(filePath: string, fileContent: Buffer | any) {
+  async upload(filePath: string, fileContent: Buffer | Readable) {
     await this.storageDriver.upload(filePath, fileContent);
     this.logger.debug(`File uploaded successfully. Path: ${filePath}`);
   }