import doc with paging

package.json

@@ -66,6 +66,7 @@
       "mongoose: somehow GlobalNotificationSetting CRUD does not work with mongoose v5.6.0",
       "openid-client: Node.js 12 or higher is required for openid-client@3 and above."
     ],
+    "JSONStream": "^1.3.5",
     "archiver": "^3.1.1",
     "async": "^3.0.1",
     "aws-sdk": "^2.88.0",

src/server/service/import.js

@@ -1,19 +1,18 @@
 const logger = require('@alias/logger')('growi:services:ImportService'); // eslint-disable-line no-unused-vars
 const fs = require('fs');
 const path = require('path');
+const JSONStream = require('JSONStream');
 const streamToPromise = require('stream-to-promise');
 
 class ImportService {
 
   constructor(crowi) {
     this.baseDir = path.join(crowi.tmpDir, 'downloads');
-    this.extension = 'json';
     this.encoding = 'utf-8';
     this.per = 100;
     this.zlibLevel = 9; // 0(min) - 9(max)
   }
 
-
   /**
    * import a collection from json
    *
@@ -22,11 +21,91 @@ class ImportService {
    * @param {string} filePath path to zipped json
    */
   async importFromZip(Model, filePath) {
-    console.log(Model.collection.collectionName);
-    console.log(filePath);
-    const readStream = fs.createReadStream(filePath);
-    const writeStream = fs.createWriteStream(path.join(this.baseDir, 'test.json'));
-    readStream.on('data', (data) => { console.log(data); writeStream.write(data); writeStream.write('---------------'); console.log('---------------') });
+    const { collectionName } = Model.collection;
+    let counter = 0;
+    let nInsertedTotal = 0;
+
+    let failedIds = [];
+    let unorderedBulkOp = Model.collection.initializeUnorderedBulkOp();
+
+
+    const readStream = fs.createReadStream(path.join(this.baseDir, 'pages.json'));
+    const jsonStream = readStream.pipe(JSONStream.parse('*'));
+
+    jsonStream.on('data', async(document) => {
+      // documents are not persisted until unorderedBulkOp.execute()
+      unorderedBulkOp.insert(document);
+
+      counter++;
+
+      if (counter % this.per === 0) {
+        // puase jsonStream to prevent more items to be added to unorderedBulkOp
+        jsonStream.pause();
+
+        const { nInserted, failed } = await this.execUnorderedBulkOpSafely(unorderedBulkOp);
+        nInsertedTotal += nInserted;
+        failedIds = [...failedIds, ...failed];
+
+        // reset initializeUnorderedBulkOp
+        unorderedBulkOp = Model.collection.initializeUnorderedBulkOp();
+
+        // resume jsonStream
+        jsonStream.resume();
+      }
+    });
+
+    jsonStream.on('end', async(data) => {
+      // insert the rest. avoid errors when unorderedBulkOp has no items
+      if (unorderedBulkOp.s.currentBatch !== null) {
+        const { nInserted, failed } = await this.execUnorderedBulkOpSafely(unorderedBulkOp);
+        nInsertedTotal += nInserted;
+        failedIds = [...failedIds, ...failed];
+      }
+
+      logger.info(`Done. Inserted ${nInsertedTotal} ${collectionName}.`);
+
+      if (failedIds.length > 0) {
+        logger.error(`Failed to insert ${failedIds.length} ${collectionName}: ${failedIds.join(', ')}.`);
+      }
+    });
+
+    // streamToPromise(jsonStream) throws error, so await readStream instead
+    await streamToPromise(readStream);
+  }
+
+  /**
+   * execute unorderedBulkOp and ignore errors
+   *
+   * @memberOf ImportService
+   * @param {object} unorderedBulkOp result of Model.collection.initializeUnorderedBulkOp()
+   * @param {string} filePath path to zipped json
+   * @return {{nInserted: number, failed: string[]}} number of docuemnts inserted and failed
+   */
+  async execUnorderedBulkOpSafely(unorderedBulkOp) {
+    // keep the number of documents inserted and failed for logger
+    let nInserted = 0;
+    const failed = [];
+
+    // try catch to skip errors
+    try {
+      const log = await unorderedBulkOp.execute();
+      nInserted = log.result.nInserted;
+    }
+    catch (err) {
+      for (const error of err.result.result.writeErrors) {
+        logger.error(error.errmsg);
+        failed.push(error.err.op._id);
+      }
+
+      nInserted = err.result.result.nInserted;
+    }
+
+    logger.debug(`Importing ${unorderedBulkOp.s.collection.s.name}. Inserted: ${nInserted}. Failed: ${failed.length}.`);
+
+    return {
+      nInserted,
+      failed,
+    };
   }
 
 }

yarn.lock

@@ -1256,6 +1256,14 @@
   resolved "https://registry.yarnpkg.com/@xtuc/long/-/long-4.2.2.tgz#d291c6a4e97989b5c61d9acf396ae4fe133a718d"
   integrity sha512-NuHqBY1PB/D8xU6s/thBgOAiAP7HOYDQ32+BFZILJ8ivkUkAHQnWfn6WhL79Owj1qmUnoN/YPhktdIoucipkAQ==
 
+JSONStream@^1.3.5:
+  version "1.3.5"
+  resolved "https://registry.yarnpkg.com/JSONStream/-/JSONStream-1.3.5.tgz#3208c1f08d3a4d99261ab64f92302bc15e111ca0"
+  integrity sha512-E+iruNOY8VV9s4JEbe1aNEm6MiszPRr/UfcHMz0TQh1BXSxHK+ASV1R6W4HpjBhSeS+54PIsAMCBmwD06LLsqQ==
+  dependencies:
+    jsonparse "^1.2.0"
+    through ">=2.2.7 <3"
+
 abab@^2.0.0:
   version "2.0.0"
   resolved "https://registry.yarnpkg.com/abab/-/abab-2.0.0.tgz#aba0ab4c5eee2d4c79d3487d85450fb2376ebb0f"
@@ -6953,6 +6961,11 @@ jsonify@~0.0.0:
   version "0.0.0"
   resolved "https://registry.yarnpkg.com/jsonify/-/jsonify-0.0.0.tgz#2c74b6ee41d93ca51b7b5aaee8f503631d252a73"
 
+jsonparse@^1.2.0:
+  version "1.3.1"
+  resolved "https://registry.yarnpkg.com/jsonparse/-/jsonparse-1.3.1.tgz#3f4dae4a91fac315f71062f8521cc239f1366280"
+  integrity sha1-P02uSpH6wxX3EGL4UhzCOfE2YoA=
+
 jsonschema-draft4@^1.0.0:
   version "1.0.0"
   resolved "https://registry.yarnpkg.com/jsonschema-draft4/-/jsonschema-draft4-1.0.0.tgz#f0af2005054f0f0ade7ea2118614b69dc512d865"
@@ -12080,7 +12093,7 @@ through2@^2.0.0:
     readable-stream "^2.1.5"
     xtend "~4.0.1"
 
-through@2, through@^2.3.6, through@~2.3, through@~2.3.1:
+through@2, "through@>=2.2.7 <3", through@^2.3.6, through@~2.3, through@~2.3.1:
   version "2.3.8"
   resolved "https://registry.yarnpkg.com/through/-/through-2.3.8.tgz#0dd4c9ffaabc357960b1b724115d7e0e86a2e1f5"