fix bad stream pipelines for fs storage backend

server/routes/upload.js

@@ -24,10 +24,14 @@ module.exports = async function(req, res) {
 
   try {
     const limiter = new Limiter(encryptedSize(config.max_file_size));
-    const fileStream = req.pipe(limiter);
     //this hasn't been updated to expiration time setting yet
     //if you want to fallback to this code add this
-    await storage.set(newId, fileStream, meta, config.default_expire_seconds);
+    await storage.set(
+      newId,
+      [req, limiter],
+      meta,
+      config.default_expire_seconds
+    );
     const url = `${config.deriveBaseUrl(req)}/download/${newId}/`;
     res.set('WWW-Authenticate', `send-v1 ${meta.nonce}`);
     res.json({

server/routes/ws.js

@@ -85,11 +85,9 @@ module.exports = function(ws, req) {
           callback();
         }
       });
-      const wsStream = ws.constructor.createWebSocketStream(ws);
+      const wsStream = ws.constructor.createWebSocketStream(ws).pipe(eof);
 
-      fileStream = wsStream.pipe(eof).pipe(limiter); // limiter needs to be the last in the chain
-
-      await storage.set(newId, fileStream, meta, timeLimit);
+      await storage.set(newId, [wsStream, limiter], meta, timeLimit);
 
       if (ws.readyState === 1) {
         // if the socket is closed by a cancelled upload the stream

server/storage/fs.js

@@ -1,5 +1,6 @@
 const fs = require('fs');
 const path = require('path');
+const { pipeline } = require('stream');
 const promisify = require('util').promisify;
 
 const stat = promisify(fs.stat);
@@ -26,15 +27,14 @@ class FSStorage {
     return new Promise((resolve, reject) => {
       const filepath = path.join(this.dir, id);
       const fstream = fs.createWriteStream(filepath);
-      file.pipe(fstream);
-      file.on('error', err => {
-        fstream.destroy(err);
-      });
-      fstream.on('error', err => {
+      pipeline([...file, fstream], err => {
+        if (err) {
           fs.unlinkSync(filepath);
           reject(err);
+        } else {
+          resolve();
+        }
       });
-      fstream.on('finish', resolve);
     });
   }