fix: use pipeline instead of pipe (#1906)

* fix: use pipeline instead of pipe

* refactor: get string without tmp file

Author: pyphilia

src/services/file/file.service.ts

@@ -6,7 +6,7 @@ import type { MaybeUser, MinimalMember } from '../../types';
 import { CachingService } from '../caching/service';
 import type { LocalFileConfiguration, S3FileConfiguration } from './interfaces/configuration';
 import type { FileRepository } from './interfaces/fileRepository';
-import { createSanitizedFile, sanitizeHtml } from './sanitize';
+import { sanitizeDocument, sanitizeHtml } from './sanitize';
 import {
   CopyFileInvalidPathError,
   CopyFolderInvalidPathError,
@@ -98,7 +98,7 @@ class FileService {
   async sanitizeFile({ file, mimetype }: { file: Readable; mimetype?: string }): Promise<Readable> {
     // sanitize content of html
     if (mimetype === 'text/html') {
-      return await createSanitizedFile(file, sanitizeHtml);
+      return await sanitizeDocument(file, sanitizeHtml);
     }
 
     return file;

src/services/file/repositories/s3.ts

@@ -11,6 +11,7 @@ import fs from 'fs';
 import { StatusCodes } from 'http-status-codes';
 import fetch from 'node-fetch';
 import path from 'path';
+import { pipeline } from 'stream/promises';
 
 import type { FastifyBaseLogger } from 'fastify';
 
@@ -194,12 +195,7 @@ export class S3FileRepository implements FileRepository {
       }
 
       const fileStream = fs.createWriteStream(filepath);
-      await new Promise<void>((resolve, reject) => {
-        res.body.pipe(fileStream);
-        res.body.on('error', reject);
-        fileStream.on('finish', resolve);
-      });
-      fileStream.end();
+      await pipeline(res.body, fileStream);
 
       // create and return read stream (similar to local file service)
       const file = fs.createReadStream(filepath);

src/services/file/sanitize.ts

@@ -1,42 +1,23 @@
-import { createWriteStream } from 'fs';
-import { readFile, unlink } from 'fs/promises';
-import path from 'path';
 import sanitize from 'sanitize-html';
 import { Readable } from 'stream';
-
-import { TMP_FOLDER } from '../../utils/config';
-import { randomHexOf4 } from '../utils';
+import { text } from 'stream/consumers';
 
 export function sanitizeHtml(content: string): string {
   return sanitize(content);
 }
 
-export function createSanitizedFile(
+export async function sanitizeDocument(
   file: Readable,
   sanitizeFn: (content: string) => string,
 ): Promise<Readable> {
-  return new Promise((done, rejects) => {
-    // create tmp file to read
-    const tmpFile = path.join(TMP_FOLDER, `${Date.now().toString()}_${randomHexOf4()}`);
-    file.pipe(createWriteStream(tmpFile));
-
-    file.on('error', function () {
-      rejects(new Error('An error happened while piping the file to sanitize'));
-    });
-
-    file.on('close', async function () {
-      const content = await readFile(tmpFile, {
-        encoding: 'utf8',
-        flag: 'r',
-      });
-
-      const readable = Readable.from(sanitizeFn(content));
-
-      // delete tmp file
-      unlink(tmpFile).catch((e) => console.error(e));
-
-      // return sanitized readable
-      done(readable);
-    });
-  });
+  try {
+    const content = await text(file);
+    const readable = Readable.from(sanitizeFn(content));
+
+    // return sanitized readable
+    return readable;
+  } catch (e) {
+    console.error(e);
+    throw new Error('An error happened while creating a sanitized version of the file');
+  }
 }

src/services/thumbnail/thumbnail.service.ts

@@ -1,6 +1,7 @@
 import path from 'path';
 import sharp from 'sharp';
 import { Readable } from 'stream';
+import { pipeline as streamPipeline } from 'stream/promises';
 import { injectable } from 'tsyringe';
 
 import { BaseLogger } from '../../logger';
@@ -40,15 +41,7 @@ export class ThumbnailService {
       Object.entries(ThumbnailSizeFormat).map(async ([sizeName, width]) => {
         // create thumbnail from image stream
         const pipeline = sharp().resize({ width }).toFormat(THUMBNAIL_FORMAT);
-        file.pipe(pipeline);
-
-        await new Promise((resolve, reject) =>
-          pipeline
-            .on('finish', () => {
-              resolve(true);
-            })
-            .on('error', reject),
-        );
+        await streamPipeline(file, pipeline);
 
         return {
           file: pipeline,