feat: Add support for streaming file upload via Buffer, Readable, ReadableStream (#10065)

Author: mtrezza

spec/GridFSBucketStorageAdapter.spec.js

@@ -476,6 +476,31 @@ describe_only_db('mongo')('GridFSBucket', () => {
     }
   });
 
+  it('reports supportsStreaming as true', () => {
+    const gfsAdapter = new GridFSBucketAdapter(databaseURI);
+    expect(gfsAdapter.supportsStreaming).toBe(true);
+  });
+
+  it('creates file from Readable stream', async () => {
+    const { Readable } = require('stream');
+    const gfsAdapter = new GridFSBucketAdapter(databaseURI);
+    const data = Buffer.from('streamed file content');
+    const stream = Readable.from(data);
+    await gfsAdapter.createFile('streamFile.txt', stream);
+    const result = await gfsAdapter.getFileData('streamFile.txt');
+    expect(result.toString('utf8')).toBe('streamed file content');
+  });
+
+  it('creates encrypted file from Readable stream (buffers for encryption)', async () => {
+    const { Readable } = require('stream');
+    const gfsAdapter = new GridFSBucketAdapter(databaseURI, {}, 'test-encryption-key');
+    const data = Buffer.from('encrypted streamed content');
+    const stream = Readable.from(data);
+    await gfsAdapter.createFile('encryptedStream.txt', stream);
+    const result = await gfsAdapter.getFileData('encryptedStream.txt');
+    expect(result.toString('utf8')).toBe('encrypted streamed content');
+  });
+
   describe('MongoDB Client Metadata', () => {
     it('should not pass metadata to MongoClient by default', async () => {
       const gfsAdapter = new GridFSBucketAdapter(databaseURI);

spec/ParseFile.spec.js

@@ -1878,4 +1878,279 @@ describe('Parse.File testing', () => {
       ).toBeRejectedWith(jasmine.objectContaining({ status: 400 }));
     });
   });
+
+  describe('streaming binary uploads', () => {
+    afterEach(() => {
+      Parse.Cloud._removeAllHooks();
+    });
+
+    describe('createSizeLimitedStream', () => {
+      const { createSizeLimitedStream } = require('../lib/Routers/FilesRouter');
+      const { Readable } = require('stream');
+
+      it('passes data through when under limit', async () => {
+        const input = Readable.from(Buffer.from('hello'));
+        const limited = createSizeLimitedStream(input, 100);
+        const chunks = [];
+        for await (const chunk of limited) {
+          chunks.push(chunk);
+        }
+        expect(Buffer.concat(chunks).toString()).toBe('hello');
+      });
+
+      it('destroys stream when data exceeds limit', async () => {
+        const input = Readable.from(Buffer.from('hello world, this is too long'));
+        const limited = createSizeLimitedStream(input, 5);
+        const chunks = [];
+        try {
+          for await (const chunk of limited) {
+            chunks.push(chunk);
+          }
+          fail('should have thrown');
+        } catch (e) {
+          expect(e.message).toContain('exceeds');
+        }
+      });
+
+    });
+
+    it('streams binary upload with X-Parse-Upload-Mode header', async () => {
+      const headers = {
+        'Content-Type': 'application/octet-stream',
+        'X-Parse-Application-Id': 'test',
+        'X-Parse-REST-API-Key': 'rest',
+        'X-Parse-Upload-Mode': 'stream',
+      };
+      let response;
+      try {
+        response = await request({
+          method: 'POST',
+          headers: headers,
+          url: 'http://localhost:8378/1/files/stream-test.txt',
+          body: 'streaming file content',
+        });
+      } catch (e) {
+        fail('Request failed: status=' + e.status + ' text=' + e.text + ' data=' + JSON.stringify(e.data));
+        return;
+      }
+      const b = response.data;
+      expect(b.name).toMatch(/_stream-test.txt$/);
+      expect(b.url).toMatch(/stream-test\.txt$/);
+      const getResponse = await request({ url: b.url });
+      expect(getResponse.text).toEqual('streaming file content');
+    });
+
+    it('infers content type from extension when Content-Type header is missing', async () => {
+      const headers = {
+        'X-Parse-Application-Id': 'test',
+        'X-Parse-REST-API-Key': 'rest',
+        'X-Parse-Upload-Mode': 'stream',
+      };
+      const response = await request({
+        method: 'POST',
+        headers: headers,
+        url: 'http://localhost:8378/1/files/inferred.txt',
+        body: 'inferred content type',
+      });
+      const b = response.data;
+      expect(b.name).toMatch(/_inferred.txt$/);
+      const getResponse = await request({ url: b.url });
+      expect(getResponse.text).toEqual('inferred content type');
+    });
+
+    it('uses buffered path without X-Parse-Upload-Mode header', async () => {
+      const headers = {
+        'Content-Type': 'application/octet-stream',
+        'X-Parse-Application-Id': 'test',
+        'X-Parse-REST-API-Key': 'rest',
+      };
+      const response = await request({
+        method: 'POST',
+        headers: headers,
+        url: 'http://localhost:8378/1/files/buffered-test.txt',
+        body: 'buffered file content',
+      });
+      const b = response.data;
+      expect(b.name).toMatch(/_buffered-test.txt$/);
+      const getResponse = await request({ url: b.url });
+      expect(getResponse.text).toEqual('buffered file content');
+    });
+
+    it('rejects streaming upload exceeding size limit', async () => {
+      await reconfigureServer({ maxUploadSize: '10b' });
+      const headers = {
+        'Content-Type': 'application/octet-stream',
+        'X-Parse-Application-Id': 'test',
+        'X-Parse-REST-API-Key': 'rest',
+        'X-Parse-Upload-Mode': 'stream',
+      };
+      try {
+        await request({
+          method: 'POST',
+          headers: headers,
+          url: 'http://localhost:8378/1/files/big-file.txt',
+          body: 'this content is definitely longer than 10 bytes',
+        });
+        fail('should have thrown');
+      } catch (response) {
+        expect(response.data.code).toBe(Parse.Error.FILE_SAVE_ERROR);
+        expect(response.data.error).toContain('exceeds');
+      }
+    });
+
+    it('rejects streaming upload with Content-Length exceeding limit', async () => {
+      await reconfigureServer({ maxUploadSize: '10b' });
+      const headers = {
+        'Content-Type': 'application/octet-stream',
+        'X-Parse-Application-Id': 'test',
+        'X-Parse-REST-API-Key': 'rest',
+        'X-Parse-Upload-Mode': 'stream',
+        'Content-Length': '99999',
+      };
+      try {
+        await request({
+          method: 'POST',
+          headers: headers,
+          url: 'http://localhost:8378/1/files/big-file.txt',
+          body: 'hi',
+        });
+        fail('should have thrown');
+      } catch (response) {
+        expect(response.data.code).toBe(Parse.Error.FILE_SAVE_ERROR);
+        expect(response.data.error).toContain('exceeds');
+      }
+    });
+
+    it('fires beforeSave trigger with request.stream = true on streaming upload', async () => {
+      let receivedStream;
+      let receivedData;
+      Parse.Cloud.beforeSave(Parse.File, (request) => {
+        receivedStream = request.stream;
+        receivedData = request.file._data;
+        request.file.addMetadata('source', 'stream');
+        request.file.addTag('env', 'test');
+      });
+      const headers = {
+        'Content-Type': 'application/octet-stream',
+        'X-Parse-Application-Id': 'test',
+        'X-Parse-REST-API-Key': 'rest',
+        'X-Parse-Upload-Mode': 'stream',
+      };
+      const response = await request({
+        method: 'POST',
+        headers: headers,
+        url: 'http://localhost:8378/1/files/trigger-test.txt',
+        body: 'trigger test content',
+      });
+      expect(response.data.name).toMatch(/_trigger-test.txt$/);
+      expect(receivedStream).toBe(true);
+      expect(receivedData).toBeFalsy();
+      const getResponse = await request({ url: response.data.url });
+      expect(getResponse.text).toEqual('trigger test content');
+    });
+
+    it('rejects streaming upload when beforeSave trigger throws', async () => {
+      Parse.Cloud.beforeSave(Parse.File, () => {
+        throw new Parse.Error(Parse.Error.SCRIPT_FAILED, 'Upload rejected');
+      });
+      const headers = {
+        'Content-Type': 'application/octet-stream',
+        'X-Parse-Application-Id': 'test',
+        'X-Parse-REST-API-Key': 'rest',
+        'X-Parse-Upload-Mode': 'stream',
+      };
+      try {
+        await request({
+          method: 'POST',
+          headers: headers,
+          url: 'http://localhost:8378/1/files/rejected.txt',
+          body: 'rejected content',
+        });
+        fail('should have thrown');
+      } catch (response) {
+        expect(response.data.code).toBe(Parse.Error.SCRIPT_FAILED);
+        expect(response.data.error).toBe('Upload rejected');
+      }
+    });
+
+    it('skips save when beforeSave trigger returns Parse.File with URL on streaming upload', async () => {
+      Parse.Cloud.beforeSave(Parse.File, () => {
+        return Parse.File.fromJSON({
+          __type: 'File',
+          name: 'existing.txt',
+          url: 'http://example.com/existing.txt',
+        });
+      });
+      const headers = {
+        'Content-Type': 'application/octet-stream',
+        'X-Parse-Application-Id': 'test',
+        'X-Parse-REST-API-Key': 'rest',
+        'X-Parse-Upload-Mode': 'stream',
+      };
+      const response = await request({
+        method: 'POST',
+        headers: headers,
+        url: 'http://localhost:8378/1/files/skip-save.txt',
+        body: 'should not be saved',
+      });
+      expect(response.data.url).toBe('http://example.com/existing.txt');
+      expect(response.data.name).toBe('existing.txt');
+    });
+
+    it('fires afterSave trigger with request.stream = true on streaming upload', async () => {
+      let afterSaveStream;
+      let afterSaveData;
+      let afterSaveUrl;
+      Parse.Cloud.afterSave(Parse.File, (request) => {
+        afterSaveStream = request.stream;
+        afterSaveData = request.file._data;
+        afterSaveUrl = request.file._url;
+      });
+      const headers = {
+        'Content-Type': 'application/octet-stream',
+        'X-Parse-Application-Id': 'test',
+        'X-Parse-REST-API-Key': 'rest',
+        'X-Parse-Upload-Mode': 'stream',
+      };
+      const response = await request({
+        method: 'POST',
+        headers: headers,
+        url: 'http://localhost:8378/1/files/after-save.txt',
+        body: 'after save content',
+      });
+      expect(response.data.name).toMatch(/_after-save.txt$/);
+      expect(afterSaveStream).toBe(true);
+      expect(afterSaveData).toBeFalsy();
+      expect(afterSaveUrl).toBeTruthy();
+    });
+
+    it('verifies FilesAdapter default supportsStreaming is false', () => {
+      const { FilesAdapter } = require('../lib/Adapters/Files/FilesAdapter');
+      const adapter = new FilesAdapter();
+      expect(adapter.supportsStreaming).toBe(false);
+    });
+
+    it('legacy JSON-wrapped upload still works', async () => {
+      await reconfigureServer({
+        fileUpload: {
+          enableForPublic: true,
+          fileExtensions: ['*'],
+        },
+      });
+      const response = await request({
+        method: 'POST',
+        url: 'http://localhost:8378/1/files/legacy.txt',
+        body: JSON.stringify({
+          _ApplicationId: 'test',
+          _JavaScriptKey: 'test',
+          _ContentType: 'text/plain',
+          base64: Buffer.from('legacy content').toString('base64'),
+        }),
+      });
+      const b = response.data;
+      expect(b.name).toMatch(/_legacy.txt$/);
+      const getResponse = await request({ url: b.url });
+      expect(getResponse.text).toEqual('legacy content');
+    });
+  });
 });

spec/Utils.spec.js

@@ -175,6 +175,81 @@ describe('Utils', () => {
     });
   });
 
+  describe('parseSizeToBytes', () => {
+    it('parses megabyte string', () => {
+      expect(Utils.parseSizeToBytes('20mb')).toBe(20 * 1024 * 1024);
+    });
+
+    it('parses Mb string (case-insensitive)', () => {
+      expect(Utils.parseSizeToBytes('20Mb')).toBe(20 * 1024 * 1024);
+    });
+
+    it('parses kilobyte string', () => {
+      expect(Utils.parseSizeToBytes('512kb')).toBe(512 * 1024);
+    });
+
+    it('parses gigabyte string', () => {
+      expect(Utils.parseSizeToBytes('1gb')).toBe(1 * 1024 * 1024 * 1024);
+    });
+
+    it('parses bytes suffix', () => {
+      expect(Utils.parseSizeToBytes('100b')).toBe(100);
+    });
+
+    it('parses plain number as bytes', () => {
+      expect(Utils.parseSizeToBytes(1048576)).toBe(1048576);
+    });
+
+    it('parses numeric string as bytes', () => {
+      expect(Utils.parseSizeToBytes('1048576')).toBe(1048576);
+    });
+
+    it('parses decimal value and floors result', () => {
+      expect(Utils.parseSizeToBytes('1.5mb')).toBe(Math.floor(1.5 * 1024 * 1024));
+    });
+
+    it('trims whitespace around value', () => {
+      expect(Utils.parseSizeToBytes('  20mb  ')).toBe(20 * 1024 * 1024);
+    });
+
+    it('allows whitespace between number and unit', () => {
+      expect(Utils.parseSizeToBytes('20 mb')).toBe(20 * 1024 * 1024);
+    });
+
+    it('parses zero', () => {
+      expect(Utils.parseSizeToBytes('0')).toBe(0);
+      expect(Utils.parseSizeToBytes(0)).toBe(0);
+    });
+
+    it('throws on invalid string', () => {
+      expect(() => Utils.parseSizeToBytes('abc')).toThrow();
+    });
+
+    it('throws on negative value', () => {
+      expect(() => Utils.parseSizeToBytes('-5mb')).toThrow();
+    });
+
+    it('throws on empty string', () => {
+      expect(() => Utils.parseSizeToBytes('')).toThrow();
+    });
+
+    it('throws on unsupported unit', () => {
+      expect(() => Utils.parseSizeToBytes('10tb')).toThrow();
+    });
+
+    it('throws on NaN', () => {
+      expect(() => Utils.parseSizeToBytes(NaN)).toThrow();
+    });
+
+    it('throws on Infinity', () => {
+      expect(() => Utils.parseSizeToBytes(Infinity)).toThrow();
+    });
+
+    it('throws on negative number', () => {
+      expect(() => Utils.parseSizeToBytes(-1)).toThrow();
+    });
+  });
+
   describe('createSanitizedError', () => {
     it('should return "Permission denied" when enableSanitizedErrorResponse is true', () => {
       const config = { enableSanitizedErrorResponse: true };