fix(server/fileuploads): deprecated endpoint now uses postgres queue (#5410)

packages/server/modules/fileuploads/rest/router.ts

@@ -1,7 +1,13 @@
 import { Router } from 'express'
-import { insertNewUploadAndNotifyFactory } from '@/modules/fileuploads/services/management'
+import {
+  insertNewUploadAndNotifyFactory,
+  insertNewUploadAndNotifyFactoryV2
+} from '@/modules/fileuploads/services/management'
 import { authMiddlewareCreator } from '@/modules/shared/middleware'
-import { saveUploadFileFactory } from '@/modules/fileuploads/repositories/fileUploads'
+import {
+  saveUploadFileFactory,
+  saveUploadFileFactoryV2
+} from '@/modules/fileuploads/repositories/fileUploads'
 import { db } from '@/db/knex'
 import { streamWritePermissionsPipelineFactory } from '@/modules/shared/authz'
 import { getStreamBranchByNameFactory } from '@/modules/core/repositories/branches'
@@ -14,6 +20,24 @@ import type { Nullable } from '@speckle/shared'
 import { ensureError } from '@speckle/shared'
 import { UploadRequestErrorMessage } from '@/modules/fileuploads/helpers/rest'
 import { getEventBus } from '@/modules/shared/services/eventBus'
+import { getFeatureFlags } from '@speckle/shared/environment'
+import { BranchNotFoundError } from '@/modules/core/errors/branch'
+import { fileImportQueues } from '@/modules/fileuploads/queues/fileimports'
+import { pushJobToFileImporterFactory } from '@/modules/fileuploads/services/createFileImport'
+import {
+  fileImportServiceShouldUsePrivateObjectsServerUrl,
+  getPrivateObjectsServerOrigin,
+  getServerOrigin
+} from '@/modules/shared/helpers/envHelper'
+import { createAppTokenFactory } from '@/modules/core/services/tokens'
+import {
+  storeApiTokenFactory,
+  storeTokenResourceAccessDefinitionsFactory,
+  storeTokenScopesFactory,
+  storeUserServerAppTokenFactory
+} from '@/modules/core/repositories/tokens'
+
+const { FF_NEXT_GEN_FILE_IMPORTER_ENABLED } = getFeatureFlags()
 
 export const fileuploadRouterFactory = (): Router => {
   const processNewFileStream = processNewFileStreamFactory()
@@ -48,11 +72,37 @@ export const fileuploadRouterFactory = (): Router => {
       const projectDb = await getProjectDbClient({ projectId })
       const getStreamBranchByName = getStreamBranchByNameFactory({ db: projectDb })
       const branch = await getStreamBranchByName(projectId, branchName)
+      if (!branch) {
+        throw new BranchNotFoundError('Branch {branchName} was not found', {
+          info: { branchName }
+        })
+      }
 
       const insertNewUploadAndNotify = insertNewUploadAndNotifyFactory({
         saveUploadFile: saveUploadFileFactory({ db: projectDb }),
         emit: getEventBus().emit
       })
+
+      const pushJobToFileImporter = pushJobToFileImporterFactory({
+        getServerOrigin: fileImportServiceShouldUsePrivateObjectsServerUrl()
+          ? getPrivateObjectsServerOrigin
+          : getServerOrigin,
+        createAppToken: createAppTokenFactory({
+          storeApiToken: storeApiTokenFactory({ db }),
+          storeTokenScopes: storeTokenScopesFactory({ db }),
+          storeTokenResourceAccessDefinitions:
+            storeTokenResourceAccessDefinitionsFactory({ db }),
+          storeUserServerAppToken: storeUserServerAppTokenFactory({ db })
+        })
+      })
+
+      const insertNewUploadAndNotifyV2 = insertNewUploadAndNotifyFactoryV2({
+        queues: fileImportQueues,
+        pushJobToFileImporter,
+        saveUploadFile: saveUploadFileFactoryV2({ db: projectDb }),
+        emit: getEventBus().emit
+      })
+
       const saveFileUploads = async ({
         uploadResults
       }: {
@@ -64,15 +114,19 @@ export const fileuploadRouterFactory = (): Router => {
       }) => {
         await Promise.all(
           uploadResults.map(async (upload) => {
-            await insertNewUploadAndNotify({
+            await (FF_NEXT_GEN_FILE_IMPORTER_ENABLED
+              ? insertNewUploadAndNotifyV2
+              : insertNewUploadAndNotify)({
               fileId: upload.blobId,
-              streamId: projectId,
-              branchName: branch?.name || branchName,
+              streamId: projectId, //legacy
+              projectId,
+              branchName: branch.name || branchName, //legacy
               userId,
               fileName: upload.fileName,
               fileType: upload.fileName?.split('.').pop() || '', //FIXME
               fileSize: upload.fileSize,
-              modelId: branch?.id || null
+              modelName: branch.name || branchName,
+              modelId: branch.id
             })
           })
         )