transloadit: fix progress with very different Assembly runtimes (#2143)

* FIX: if an assembly has fully completed during the upload step (because it was smaller than other files in the batch), complete it during post processing

* FIX: linter error

* TRY: creating AssemblyWatcher before connecting assembly

* REMOVE: Assembly watcher changes

* transloadit: track completed files by ID

* transloadit: fix postprocess progress bar

* transloadit: create AssemblyWatchers when restoring files

Co-authored-by: Renée Kooi <renee@kooi.me>

packages/@uppy/transloadit/src/index.js

@@ -1,4 +1,5 @@
 const Translator = require('@uppy/utils/lib/Translator')
+const hasProperty = require('@uppy/utils/lib/hasProperty')
 const { Plugin } = require('@uppy/core')
 const Tus = require('@uppy/tus')
 const Assembly = require('./Assembly')
@@ -84,6 +85,10 @@ module.exports = class Transloadit extends Plugin {
     })
     // Contains Assembly instances for in-progress Assemblies.
     this.activeAssemblies = {}
+    // Contains a mapping of uploadID to AssemblyWatcher
+    this.assemblyWatchers = {}
+    // Contains a file IDs that have completed postprocessing before the upload they belong to has entered the postprocess stage.
+    this.completedFiles = Object.create(null)
   }
 
   setOptions (newOpts) {
@@ -208,20 +213,21 @@ module.exports = class Transloadit extends Plugin {
     }).then((newAssembly) => {
       const assembly = new Assembly(newAssembly)
       const status = assembly.status
+      const assemblyID = status.assembly_id
 
       const { assemblies, uploadsAssemblies } = this.getPluginState()
       this.setPluginState({
         // Store the Assembly status.
         assemblies: {
           ...assemblies,
-          [status.assembly_id]: status
+          [assemblyID]: status
         },
         // Store the list of Assemblies related to this upload.
         uploadsAssemblies: {
           ...uploadsAssemblies,
           [uploadID]: [
             ...uploadsAssemblies[uploadID],
-            status.assembly_id
+            assemblyID
           ]
         }
       })
@@ -240,9 +246,7 @@ module.exports = class Transloadit extends Plugin {
 
       this.uppy.emit('transloadit:assembly-created', status, fileIDs)
 
-      this._connectAssembly(assembly)
-
-      this.uppy.log(`[Transloadit] Created Assembly ${status.assembly_id}`)
+      this.uppy.log(`[Transloadit] Created Assembly ${assemblyID}`)
       return assembly
     }).catch((err) => {
       err.message = `${this.i18n('creatingAssemblyFailed')}: ${err.message}`
@@ -251,6 +255,32 @@ module.exports = class Transloadit extends Plugin {
     })
   }
 
+  _createAssemblyWatcher (assemblyID, fileIDs, uploadID) {
+  // AssemblyWatcher tracks completion states of all Assemblies in this upload.
+    const watcher = new AssemblyWatcher(this.uppy, assemblyID)
+
+    watcher.on('assembly-complete', (id) => {
+      const files = this.getAssemblyFiles(id)
+      files.forEach((file) => {
+        this.completedFiles[file.id] = true
+        this.uppy.emit('postprocess-complete', file)
+      })
+    })
+
+    watcher.on('assembly-error', (id, error) => {
+    // Clear postprocessing state for all our files.
+      const files = this.getAssemblyFiles(id)
+      files.forEach((file) => {
+      // TODO Maybe make a postprocess-error event here?
+        this.uppy.emit('upload-error', file, error)
+
+        this.uppy.emit('postprocess-complete', file)
+      })
+    })
+
+    this.assemblyWatchers[uploadID] = watcher
+  }
+
   _shouldWaitAfterUpload () {
     return this.opts.waitForEncoding || this.opts.waitForMetadata
   }
@@ -360,11 +390,12 @@ module.exports = class Transloadit extends Plugin {
   _onAssemblyFinished (status) {
     const url = status.assembly_ssl_url
     this.client.getAssemblyStatus(url).then((finalStatus) => {
+      const assemblyId = finalStatus.assemblyId
       const state = this.getPluginState()
       this.setPluginState({
         assemblies: {
           ...state.assemblies,
-          [finalStatus.assembly_id]: finalStatus
+          [assemblyId]: finalStatus
         }
       })
       this.uppy.emit('transloadit:complete', finalStatus)
@@ -462,10 +493,23 @@ module.exports = class Transloadit extends Plugin {
       })
     }
 
-    // Set up the Assembly instances for existing Assemblies.
+    // Set up the Assembly instances and AssemblyWatchers for existing Assemblies.
     const restoreAssemblies = () => {
-      const { assemblies } = this.getPluginState()
-      Object.keys(assemblies).forEach((id) => {
+      const { assemblies, uploadsAssemblies } = this.getPluginState()
+
+      // Set up the assembly watchers again for all the ongoing uploads.
+      Object.keys(uploadsAssemblies).forEach((uploadID) => {
+        const assemblyIDs = uploadsAssemblies[uploadID]
+        const fileIDsInUpload = assemblyIDs.reduce((acc, assemblyID) => {
+          const fileIDsInAssembly = this.getAssemblyFiles(assemblyID).map((file) => file.id)
+          acc.push(...fileIDsInAssembly)
+          return acc
+        }, [])
+        this._createAssemblyWatcher(assemblyIDs, fileIDsInUpload, uploadID)
+      })
+
+      const allAssemblyIDs = Object.keys(assemblies)
+      allAssemblyIDs.forEach((id) => {
         const assembly = new Assembly(assemblies[id])
         this._connectAssembly(assembly)
       })
@@ -569,7 +613,9 @@ module.exports = class Transloadit extends Plugin {
     })
 
     const createAssembly = ({ fileIDs, options }) => {
+      let createdAssembly
       return this._createAssembly(fileIDs, uploadID, options).then((assembly) => {
+        createdAssembly = assembly
         if (this.opts.importFromUploadURLs) {
           return this._reserveFiles(assembly, fileIDs)
         }
@@ -578,6 +624,7 @@ module.exports = class Transloadit extends Plugin {
           const file = this.uppy.getFile(fileID)
           this.uppy.emit('preprocess-complete', file)
         })
+        return createdAssembly
       }).catch((err) => {
         fileIDs.forEach((fileID) => {
           const file = this.uppy.getFile(fileID)
@@ -604,7 +651,11 @@ module.exports = class Transloadit extends Plugin {
     return assemblyOptions.build().then(
       (assemblies) => Promise.all(
         assemblies.map(createAssembly)
-      ),
+      ).then((createdAssemblies) => {
+        const assemblyIDs = createdAssemblies.map(assembly => assembly.status.assembly_id)
+        this._createAssemblyWatcher(assemblyIDs, fileIDs, uploadID)
+        createdAssemblies.map(assembly => this._connectAssembly(assembly))
+      }),
       // If something went wrong before any Assemblies could be created,
       // clear all processing state.
       (err) => {
@@ -619,8 +670,9 @@ module.exports = class Transloadit extends Plugin {
   }
 
   _afterUpload (fileIDs, uploadID) {
+    const files = fileIDs.map(fileID => this.uppy.getFile(fileID))
     // Only use files without errors
-    fileIDs = fileIDs.filter((file) => !file.error)
+    fileIDs = files.filter((file) => !file.error).map(file => file.id)
 
     const state = this.getPluginState()
 
@@ -653,35 +705,15 @@ module.exports = class Transloadit extends Plugin {
       return Promise.resolve()
     }
 
-    // AssemblyWatcher tracks completion states of all Assemblies in this upload.
-    const watcher = new AssemblyWatcher(this.uppy, assemblyIDs)
-
-    fileIDs.forEach((fileID) => {
-      const file = this.uppy.getFile(fileID)
+    const incompleteFiles = files.filter(file => !hasProperty(this.completedFiles, file.id))
+    incompleteFiles.forEach((file) => {
       this.uppy.emit('postprocess-progress', file, {
         mode: 'indeterminate',
         message: this.i18n('encoding')
       })
     })
 
-    watcher.on('assembly-complete', (id) => {
-      const files = this.getAssemblyFiles(id)
-      files.forEach((file) => {
-        this.uppy.emit('postprocess-complete', file)
-      })
-    })
-
-    watcher.on('assembly-error', (id, error) => {
-      // Clear postprocessing state for all our files.
-      const files = this.getAssemblyFiles(id)
-      files.forEach((file) => {
-        // TODO Maybe make a postprocess-error event here?
-        this.uppy.emit('upload-error', file, error)
-
-        this.uppy.emit('postprocess-complete', file)
-      })
-    })
-
+    const watcher = this.assemblyWatchers[uploadID]
     return watcher.promise.then(() => {
       const assemblies = assemblyIDs.map((id) => this.getAssembly(id))