perf(import): stream excel file import

Close erxes/erxes#2075

package.json

@@ -94,6 +94,7 @@
     "uuid-by-string": "^3.0.2",
     "validator": "^9.0.0",
     "xlsx-populate": "^1.20.1",
+    "xlsx-stream-reader": "^1.1.0",
     "xss": "^1.0.6"
   },
   "peerOptionalDependencies": {

src/data/utils.ts

@@ -124,8 +124,11 @@ const createGCS = async () => {
 /*
  * Save binary data to amazon s3
  */
-export const uploadFileAWS = async (file: { name: string; path: string; type: string }): Promise<string> => {
-  const IS_PUBLIC = await getConfig('FILE_SYSTEM_PUBLIC', 'true');
+export const uploadFileAWS = async (
+  file: { name: string; path: string; type: string },
+  forcePrivate: boolean = false,
+): Promise<string> => {
+  const IS_PUBLIC = forcePrivate ? false : await getConfig('FILE_SYSTEM_PUBLIC', 'true');
   const AWS_PREFIX = await getConfig('AWS_PREFIX');
   const AWS_BUCKET = await getConfig('AWS_BUCKET');
 
@@ -164,7 +167,7 @@ export const uploadFileAWS = async (file: { name: string; path: string; type: st
 /*
  * Delete file from amazon s3
  */
-const deleteFileAWS = async (fileName: string) => {
+export const deleteFileAWS = async (fileName: string) => {
   const AWS_BUCKET = await getConfig('AWS_BUCKET');
 
   const params = { Bucket: AWS_BUCKET, Key: fileName };
@@ -992,3 +995,18 @@ export const chunkArray = (myArray, chunkSize: number) => {
 
   return tempArray;
 };
+
+/**
+ * Create s3 stream for excel file
+ */
+export const s3Stream = async (key: string): Promise<any> => {
+  try {
+    const AWS_BUCKET = await getConfig('AWS_BUCKET');
+
+    const s3 = await createAWS();
+
+    return s3.getObject({ Bucket: AWS_BUCKET, Key: key }).createReadStream();
+  } catch (e) {
+    throw e;
+  }
+};

src/middlewares/fileMiddleware.ts

@@ -1,74 +1,12 @@
 import * as formidable from 'formidable';
-import * as fs from 'fs';
 import * as request from 'request';
 import * as _ from 'underscore';
-import * as xlsxPopulate from 'xlsx-populate';
 import { RABBITMQ_QUEUES } from '../data/constants';
 import { can } from '../data/permissions/utils';
-import { checkFile, frontendEnv, getSubServiceDomain, uploadFile } from '../data/utils';
+import { checkFile, frontendEnv, getSubServiceDomain, uploadFile, uploadFileAWS } from '../data/utils';
 import { debugExternalApi } from '../debuggers';
 import { sendRPCMessage } from '../messageBroker';
 
-const readXlsFile = async (file): Promise<{ fieldNames: string[]; usedSheets: any[] }> => {
-  return new Promise(async (resolve, reject) => {
-    const readStream = fs.createReadStream(file.path);
-
-    // Directory to save file
-    const downloadDir = `${__dirname}/../private/xlsTemplateOutputs/${file.name}`;
-
-    // Converting pipe into promise
-    const pipe = stream =>
-      new Promise((resolver, rejecter) => {
-        stream.on('finish', resolver);
-        stream.on('error', rejecter);
-      });
-
-    // Creating streams
-    const writeStream = fs.createWriteStream(downloadDir);
-    const streamObj = await readStream.pipe(writeStream);
-
-    pipe(streamObj)
-      .then(async () => {
-        // After finished saving instantly create and load workbook from xls
-        const workbook = await xlsxPopulate.fromFileAsync(downloadDir);
-
-        // Deleting file after read
-        fs.unlink(downloadDir, () => {
-          return true;
-        });
-
-        const usedRange = workbook.sheet(0).usedRange();
-
-        if (!usedRange) {
-          return reject(new Error('Invalid file'));
-        }
-
-        const usedSheets = usedRange.value();
-        const compactedRows: any[] = [];
-
-        for (const row of usedSheets) {
-          // to prevent empty data entry since xlsPopulate returns empty cells
-          const compactRow = _.compact(row);
-
-          if (compactRow.length > 0) {
-            compactedRows.push(row);
-          }
-        }
-
-        // Getting columns
-        const fieldNames = usedSheets[0];
-
-        // Removing column
-        compactedRows.shift();
-
-        return resolve({ fieldNames, usedSheets: compactedRows });
-      })
-      .catch(e => {
-        return reject(e);
-      });
-  });
-};
-
 export const importer = async (req: any, res, next) => {
   if (!(await can('importXlsFile', req.user))) {
     return next(new Error('Permission denied!'));
@@ -93,13 +31,12 @@ export const importer = async (req: any, res, next) => {
       }
 
       try {
-        const { fieldNames, usedSheets } = await readXlsFile(response.file);
+        const fileName = await uploadFileAWS(response.file, true);
 
         const result = await sendRPCMessage(RABBITMQ_QUEUES.RPC_API_TO_WORKERS, {
           action: 'createImport',
           type: fields.type,
-          fieldNames,
-          datas: usedSheets,
+          fileName,
           scopeBrandIds,
           user: req.user,
         });

src/workers/bulkInsert.worker.ts

@@ -14,6 +14,7 @@ import {
   Tickets,
   Users,
 } from '../db/models';
+import { initRabbitMQ } from '../messageBroker';
 import { graphqlPubsub } from '../pubsub';
 import { connect } from './utils';
 
@@ -34,6 +35,8 @@ connect().then(async () => {
     return;
   }
 
+  await initRabbitMQ();
+
   const { user, scopeBrandIds, result, contentType, properties, importHistoryId, percentagePerData } = workerData;
 
   let percentage = '0';
@@ -198,6 +201,10 @@ connect().then(async () => {
       doc.emailValidationStatus = 'unknown';
     }
 
+    if (contentType === 'customer' && !doc.phoneValidationStatus) {
+      doc.phoneValidationStatus = 'unknown';
+    }
+
     // set board item created user
     if (isBoardItem()) {
       doc.userId = user._id;

src/workers/utils.ts

@@ -2,12 +2,15 @@ import * as dotenv from 'dotenv';
 import * as mongoose from 'mongoose';
 import * as os from 'os';
 import * as path from 'path';
+import * as XlsxStreamReader from 'xlsx-stream-reader';
 import { checkFieldNames } from '../data/modules/fields/utils';
+import { deleteFileAWS, s3Stream } from '../data/utils';
 import { ImportHistory } from '../db/models';
 import ImportHistories from '../db/models/ImportHistory';
 import { debugImport } from '../debuggers';
 
 const { MONGO_URL = '' } = process.env;
+
 export const connect = () => mongoose.connect(MONGO_URL, { useNewUrlParser: true, useCreateIndex: true });
 
 dotenv.config();
@@ -130,8 +133,70 @@ export const receiveImportCancel = () => {
   return { status: 'ok' };
 };
 
+const readXlsFile = async (fileName: string): Promise<{ fieldNames: string[]; datas: any[] }> => {
+  return new Promise(async (resolve, reject) => {
+    let rowCount = 0;
+
+    const usedSheets: any[] = [];
+
+    const xlsxReader = XlsxStreamReader();
+
+    try {
+      const stream = await s3Stream(fileName);
+
+      stream.pipe(xlsxReader);
+
+      xlsxReader.on('worksheet', workSheetReader => {
+        if (workSheetReader > 1) {
+          return workSheetReader.skip();
+        }
+
+        workSheetReader.on('row', row => {
+          if (rowCount > 100000) {
+            return reject(new Error('You can only import 100000 rows one at a time'));
+          }
+
+          if (row.values.length > 0) {
+            usedSheets.push(row.values);
+            rowCount++;
+          }
+        });
+
+        workSheetReader.process();
+      });
+
+      xlsxReader.on('end', () => {
+        const compactedRows: any = [];
+
+        for (const row of usedSheets) {
+          if (row.length > 0) {
+            row.shift();
+
+            compactedRows.push(row);
+          }
+        }
+
+        const fieldNames = usedSheets[0];
+
+        // Removing column
+        compactedRows.shift();
+
+        return resolve({ fieldNames, datas: compactedRows });
+      });
+
+      xlsxReader.on('error', error => {
+        return reject(error);
+      });
+    } catch (e) {
+      reject(e);
+    }
+  });
+};
+
 export const receiveImportXls = async (content: any) => {
-  const { type, fieldNames, scopeBrandIds, user, datas } = content;
+  const { fileName, type, scopeBrandIds, user } = content;
+
+  const { fieldNames, datas } = await readXlsFile(fileName);
 
   if (datas.length === 0) {
     throw new Error('Please import at least one row of data');
@@ -168,5 +233,7 @@ export const receiveImportXls = async (content: any) => {
 
   await createWorkers(workerPath, workerData, results);
 
+  await deleteFileAWS(fileName);
+
   return { id: importHistory.id };
 };