@uppy/aws-s3-multipart: refresh file before calling user-defined functions (#4557)

- https://community.transloadit.com/t/custom-file-metadata-not-persisted-across-user-defined-functions-in-aws-s3-multipart/16661/3
- https://github.com/transloadit/uppy/issues/4533

Co-authored-by: Madeline Lumetta <mlumetta@loc.gov>

packages/@uppy/aws-s3-multipart/src/index.js

@@ -86,9 +86,12 @@ class HTTPCommunicationQueue {
 
   #uploadPartBytes
 
-  constructor (requests, options, setS3MultipartState) {
+  #getFile
+
+  constructor (requests, options, setS3MultipartState, getFile) {
     this.#requests = requests
     this.#setS3MultipartState = setS3MultipartState
+    this.#getFile = getFile
     this.setOptions(options)
   }
 
@@ -197,7 +200,7 @@ class HTTPCommunicationQueue {
       }
     }
 
-    const promise = this.#createMultipartUpload(file, signal)
+    const promise = this.#createMultipartUpload(this.#getFile(file), signal)
 
     const abortPromise = () => {
       promise.abort(signal.reason)
@@ -235,7 +238,7 @@ class HTTPCommunicationQueue {
       // If the cached result rejects, there's nothing to abort.
       return
     }
-    await this.#abortMultipartUpload(file, awaitedResult)
+    await this.#abortMultipartUpload(this.#getFile(file), awaitedResult)
   }
 
   async #nonMultipartUpload (file, chunk, signal) {
@@ -244,7 +247,7 @@ class HTTPCommunicationQueue {
       url,
       fields,
       headers,
-    } = await this.#getUploadParameters(file, { signal }).abortOn(signal)
+    } = await this.#getUploadParameters(this.#getFile(file), { signal }).abortOn(signal)
 
     let body
     const data = chunk.getData()
@@ -285,7 +288,10 @@ class HTTPCommunicationQueue {
     try {
       const parts = await Promise.all(chunks.map((chunk, i) => this.uploadChunk(file, i + 1, chunk, signal)))
       throwIfAborted(signal)
-      return await this.#sendCompletionRequest(file, { key, uploadId, parts, signal }).abortOn(signal)
+      return await this.#sendCompletionRequest(
+        this.#getFile(file),
+        { key, uploadId, parts, signal },
+      ).abortOn(signal)
     } catch (err) {
       if (err?.cause !== pausingUploadReason && err?.name !== 'AbortError') {
         // We purposefully don't wait for the promise and ignore its status,
@@ -308,7 +314,10 @@ class HTTPCommunicationQueue {
     }
     const { uploadId, key } = await this.getUploadId(file, signal)
     throwIfAborted(signal)
-    const alreadyUploadedParts = await this.#listParts(file, { uploadId, key, signal }).abortOn(signal)
+    const alreadyUploadedParts = await this.#listParts(
+      this.#getFile(file),
+      { uploadId, key, signal },
+    ).abortOn(signal)
     throwIfAborted(signal)
     const parts = await Promise.all(
       chunks
@@ -324,7 +333,10 @@ class HTTPCommunicationQueue {
         }),
     )
     throwIfAborted(signal)
-    return this.#sendCompletionRequest(file, { key, uploadId, parts, signal }).abortOn(signal)
+    return this.#sendCompletionRequest(
+      this.#getFile(file),
+      { key, uploadId, parts, signal },
+    ).abortOn(signal)
   }
 
   /**
@@ -343,7 +355,7 @@ class HTTPCommunicationQueue {
       const chunkData = chunk.getData()
       const { onProgress, onComplete } = chunk
 
-      const signature = await this.#fetchSignature(file, {
+      const signature = await this.#fetchSignature(this.#getFile(file), {
         uploadId, key, partNumber, body: chunkData, signal,
       }).abortOn(signal)
 
@@ -413,7 +425,12 @@ export default class AwsS3Multipart extends UploaderPlugin {
      * @type {RateLimitedQueue}
      */
     this.requests = this.opts.rateLimitedQueue ?? new RateLimitedQueue(this.opts.limit)
-    this.#companionCommunicationQueue = new HTTPCommunicationQueue(this.requests, this.opts, this.#setS3MultipartState)
+    this.#companionCommunicationQueue = new HTTPCommunicationQueue(
+      this.requests,
+      this.opts,
+      this.#setS3MultipartState,
+      this.#getFile,
+    )
 
     this.uploaders = Object.create(null)
     this.uploaderEvents = Object.create(null)
@@ -675,6 +692,11 @@ export default class AwsS3Multipart extends UploaderPlugin {
 
   #setS3MultipartState = (file, { key, uploadId }) => {
     const cFile = this.uppy.getFile(file.id)
+    if (cFile == null) {
+      // file was removed from store
+      return
+    }
+
     this.uppy.setFileState(file.id, {
       s3Multipart: {
         ...cFile.s3Multipart,
@@ -684,10 +706,12 @@ export default class AwsS3Multipart extends UploaderPlugin {
     })
   }
 
+  #getFile = (file) => {
+    return this.uppy.getFile(file.id) || file
+  }
+
   #uploadFile (file) {
     return new Promise((resolve, reject) => {
-      const getFile = () => this.uppy.getFile(file.id) || file
-
       const onProgress = (bytesUploaded, bytesTotal) => {
         this.uppy.emit('upload-progress', file, {
           uploader: this,
@@ -714,7 +738,7 @@ export default class AwsS3Multipart extends UploaderPlugin {
 
         this.resetUploaderReferences(file.id)
 
-        this.uppy.emit('upload-success', getFile(), uploadResp)
+        this.uppy.emit('upload-success', this.#getFile(file), uploadResp)
 
         if (result.location) {
           this.uppy.log(`Download ${file.name} from ${result.location}`)
@@ -724,7 +748,7 @@ export default class AwsS3Multipart extends UploaderPlugin {
       }
 
       const onPartComplete = (part) => {
-        this.uppy.emit('s3-multipart:part-uploaded', getFile(), part)
+        this.uppy.emit('s3-multipart:part-uploaded', this.#getFile(file), part)
       }
 
       const upload = new MultipartUploader(file.data, {

packages/@uppy/aws-s3-multipart/src/index.test.js

@@ -370,4 +370,239 @@ describe('AwsS3Multipart', () => {
       expect(client[Symbol.for('uppy test: getCompanionHeaders')]().authorization).toEqual(newToken)
     })
   })
+
+  describe('file metadata across custom main functions', () => {
+    let core
+    const createMultipartUpload = jest.fn(file => {
+      core.setFileMeta(file.id, {
+        ...file.meta,
+        createMultipartUpload: true,
+      })
+
+      return {
+        uploadId: 'upload1234',
+        key: file.name,
+      }
+    })
+
+    const signPart = jest.fn((file, partData) => {
+      expect(file.meta.createMultipartUpload).toBe(true)
+
+      core.setFileMeta(file.id, {
+        ...file.meta,
+        signPart: true,
+        [`part${partData.partNumber}`]: partData.partNumber,
+      })
+
+      return {
+        url: `https://bucket.s3.us-east-2.amazonaws.com/test/upload/multitest.dat?partNumber=${partData.partNumber}&uploadId=6aeb1980f3fc7ce0b5454d25b71992&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKIATEST%2F20210729%2Fus-east-2%2Fs3%2Faws4_request&X-Amz-Date=20210729T014044Z&X-Amz-Expires=600&X-Amz-SignedHeaders=host&X-Amz-Signature=test`,
+      }
+    })
+
+    const listParts = jest.fn((file) => {
+      expect(file.meta.createMultipartUpload).toBe(true)
+      core.setFileMeta(file.id, {
+        ...file.meta,
+        listParts: true,
+      })
+
+      const partKeys = Object.keys(file.meta).filter(metaKey => metaKey.startsWith('part'))
+      return partKeys.map(metaKey => ({
+        PartNumber: file.meta[metaKey],
+        ETag: metaKey,
+        Size: 5 * MB,
+      }))
+    })
+
+    const completeMultipartUpload = jest.fn((file) => {
+      expect(file.meta.createMultipartUpload).toBe(true)
+      expect(file.meta.signPart).toBe(true)
+      for (let i = 1; i <= 10; i++) {
+        expect(file.meta[`part${i}`]).toBe(i)
+      }
+      return {}
+    })
+
+    const abortMultipartUpload = jest.fn((file) => {
+      expect(file.meta.createMultipartUpload).toBe(true)
+      expect(file.meta.signPart).toBe(true)
+      expect(file.meta.abortingPart).toBe(5)
+      return {}
+    })
+
+    beforeEach(() => {
+      createMultipartUpload.mockClear()
+      signPart.mockClear()
+      listParts.mockClear()
+      abortMultipartUpload.mockClear()
+      completeMultipartUpload.mockClear()
+    })
+
+    it('preserves file metadata if upload is completed', async () => {
+      core = new Core()
+        .use(AwsS3Multipart, {
+          createMultipartUpload,
+          signPart,
+          listParts,
+          completeMultipartUpload,
+          abortMultipartUpload,
+        })
+
+      nock('https://bucket.s3.us-east-2.amazonaws.com')
+        .defaultReplyHeaders({
+          'access-control-allow-headers': '*',
+          'access-control-allow-method': 'PUT',
+          'access-control-allow-origin': '*',
+          'access-control-expose-headers': 'ETag',
+        })
+        .put((uri) => uri.includes('test/upload/multitest.dat'))
+        .reply(200, '', { ETag: 'test' })
+        .persist()
+
+      const fileSize = 50 * MB
+      core.addFile({
+        source: 'jest',
+        name: 'multitest.dat',
+        type: 'application/octet-stream',
+        data: new File([new Uint8Array(fileSize)], {
+          type: 'application/octet-stream',
+        }),
+      })
+
+      await core.upload()
+      expect(createMultipartUpload).toHaveBeenCalled()
+      expect(signPart).toHaveBeenCalledTimes(10)
+      expect(completeMultipartUpload).toHaveBeenCalled()
+    })
+
+    it('preserves file metadata if upload is aborted', async () => {
+      const signPartWithAbort = jest.fn((file, partData) => {
+        expect(file.meta.createMultipartUpload).toBe(true)
+        if (partData.partNumber === 5) {
+          core.setFileMeta(file.id, {
+            ...file.meta,
+            abortingPart: partData.partNumber,
+          })
+          core.removeFile(file.id)
+          return {}
+        }
+
+        core.setFileMeta(file.id, {
+          ...file.meta,
+          signPart: true,
+          [`part${partData.partNumber}`]: partData.partNumber,
+        })
+
+        return {
+          url: `https://bucket.s3.us-east-2.amazonaws.com/test/upload/multitest.dat?partNumber=${partData.partNumber}&uploadId=6aeb1980f3fc7ce0b5454d25b71992&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKIATEST%2F20210729%2Fus-east-2%2Fs3%2Faws4_request&X-Amz-Date=20210729T014044Z&X-Amz-Expires=600&X-Amz-SignedHeaders=host&X-Amz-Signature=test`,
+        }
+      })
+
+      core = new Core()
+        .use(AwsS3Multipart, {
+          createMultipartUpload,
+          signPart: signPartWithAbort,
+          listParts,
+          completeMultipartUpload,
+          abortMultipartUpload,
+        })
+
+      nock('https://bucket.s3.us-east-2.amazonaws.com')
+        .defaultReplyHeaders({
+          'access-control-allow-headers': '*',
+          'access-control-allow-method': 'PUT',
+          'access-control-allow-origin': '*',
+          'access-control-expose-headers': 'ETag',
+        })
+        .put((uri) => uri.includes('test/upload/multitest.dat'))
+        .reply(200, '', { ETag: 'test' })
+        .persist()
+
+      const fileSize = 50 * MB
+      core.addFile({
+        source: 'jest',
+        name: 'multitest.dat',
+        type: 'application/octet-stream',
+        data: new File([new Uint8Array(fileSize)], {
+          type: 'application/octet-stream',
+        }),
+      })
+
+      await core.upload()
+      expect(createMultipartUpload).toHaveBeenCalled()
+      expect(signPartWithAbort).toHaveBeenCalled()
+      expect(abortMultipartUpload).toHaveBeenCalled()
+    })
+
+    it('preserves file metadata if upload is paused and resumed', async () => {
+      const completeMultipartUploadAfterPause = jest.fn((file) => {
+        expect(file.meta.createMultipartUpload).toBe(true)
+        expect(file.meta.signPart).toBe(true)
+        for (let i = 1; i <= 10; i++) {
+          expect(file.meta[`part${i}`]).toBe(i)
+        }
+
+        expect(file.meta.listParts).toBe(true)
+        return {}
+      })
+
+      const signPartWithPause = jest.fn((file, partData) => {
+        expect(file.meta.createMultipartUpload).toBe(true)
+        if (partData.partNumber === 3) {
+          core.setFileMeta(file.id, {
+            ...file.meta,
+            abortingPart: partData.partNumber,
+          })
+          core.pauseResume(file.id)
+          setTimeout(() => core.pauseResume(file.id), 500)
+        }
+
+        core.setFileMeta(file.id, {
+          ...file.meta,
+          signPart: true,
+          [`part${partData.partNumber}`]: partData.partNumber,
+        })
+
+        return {
+          url: `https://bucket.s3.us-east-2.amazonaws.com/test/upload/multitest.dat?partNumber=${partData.partNumber}&uploadId=6aeb1980f3fc7ce0b5454d25b71992&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKIATEST%2F20210729%2Fus-east-2%2Fs3%2Faws4_request&X-Amz-Date=20210729T014044Z&X-Amz-Expires=600&X-Amz-SignedHeaders=host&X-Amz-Signature=test`,
+        }
+      })
+
+      core = new Core()
+        .use(AwsS3Multipart, {
+          createMultipartUpload,
+          signPart: signPartWithPause,
+          listParts,
+          completeMultipartUpload: completeMultipartUploadAfterPause,
+          abortMultipartUpload,
+        })
+
+      nock('https://bucket.s3.us-east-2.amazonaws.com')
+        .defaultReplyHeaders({
+          'access-control-allow-headers': '*',
+          'access-control-allow-method': 'PUT',
+          'access-control-allow-origin': '*',
+          'access-control-expose-headers': 'ETag',
+        })
+        .put((uri) => uri.includes('test/upload/multitest.dat'))
+        .reply(200, '', { ETag: 'test' })
+        .persist()
+
+      const fileSize = 50 * MB
+      core.addFile({
+        source: 'jest',
+        name: 'multitest.dat',
+        type: 'application/octet-stream',
+        data: new File([new Uint8Array(fileSize)], {
+          type: 'application/octet-stream',
+        }),
+      })
+
+      await core.upload()
+      expect(createMultipartUpload).toHaveBeenCalled()
+      expect(signPartWithPause).toHaveBeenCalled()
+      expect(listParts).toHaveBeenCalled()
+      expect(completeMultipartUploadAfterPause).toHaveBeenCalled()
+    })
+  })
 })