Merge pull request #3320 from weseek/feature/4685-4906-create-rename-stream

Feature/4685 4906 create rename stream

src/server/models/page.js

@@ -1133,103 +1133,6 @@ module.exports = function(crowi) {
     await this.removeRedirectOriginPageByPath(redirectPage.path);
   };
 
-  pageSchema.statics.rename = async function(pageData, newPagePath, user, options) {
-    validateCrowi();
-
-    const Page = this;
-    const Revision = crowi.model('Revision');
-    const path = pageData.path;
-    const createRedirectPage = options.createRedirectPage || false;
-    const updateMetadata = options.updateMetadata || false;
-    const socketClientId = options.socketClientId || null;
-
-    // sanitize path
-    newPagePath = crowi.xss.process(newPagePath); // eslint-disable-line no-param-reassign
-
-    // update Page
-    pageData.path = newPagePath;
-    if (updateMetadata) {
-      pageData.lastUpdateUser = user;
-      pageData.updatedAt = Date.now();
-    }
-    const updatedPageData = await pageData.save();
-
-    // update Rivisions
-    await Revision.updateRevisionListByPath(path, { path: newPagePath }, {});
-
-    if (createRedirectPage) {
-      const body = `redirect ${newPagePath}`;
-      await Page.create(path, body, user, { redirectTo: newPagePath });
-    }
-
-    pageEvent.emit('delete', pageData, user, socketClientId);
-    pageEvent.emit('create', updatedPageData, user, socketClientId);
-
-    return updatedPageData;
-  };
-
-  pageSchema.statics.renameRecursively = async function(targetPage, newPagePathPrefix, user, options) {
-    validateCrowi();
-
-    const pageCollection = mongoose.connection.collection('pages');
-    const revisionCollection = mongoose.connection.collection('revisions');
-
-    const path = targetPage.path;
-    const pathRegExp = new RegExp(`^${escapeStringRegexp(path)}`, 'i');
-    const { updateMetadata, createRedirectPage } = options;
-
-    // sanitize path
-    newPagePathPrefix = crowi.xss.process(newPagePathPrefix); // eslint-disable-line no-param-reassign
-
-    // find manageable descendants
-    const pages = await this.findManageableListWithDescendants(targetPage, user, options);
-
-    const unorderedBulkOp = pageCollection.initializeUnorderedBulkOp();
-    const createRediectPageBulkOp = pageCollection.initializeUnorderedBulkOp();
-    const revisionUnorderedBulkOp = revisionCollection.initializeUnorderedBulkOp();
-
-    pages.forEach((page) => {
-      const newPagePath = page.path.replace(pathRegExp, newPagePathPrefix);
-      if (updateMetadata) {
-        unorderedBulkOp.find({ _id: page._id }).update([{ $set: { path: newPagePath, lastUpdateUser: user._id, updatedAt: { $toDate: Date.now() } } }]);
-      }
-      else {
-        unorderedBulkOp.find({ _id: page._id }).update({ $set: { path: newPagePath } });
-      }
-      if (createRedirectPage) {
-        createRediectPageBulkOp.insert({
-          path: page.path, body: `redirect ${newPagePath}`, creator: user, lastUpdateUser: user, status: STATUS_PUBLISHED, redirectTo: newPagePath,
-        });
-      }
-      revisionUnorderedBulkOp.find({ path: page.path }).update({ $set: { path: newPagePath } }, { multi: true });
-    });
-
-    try {
-      await unorderedBulkOp.execute();
-      await revisionUnorderedBulkOp.execute();
-    }
-    catch (err) {
-      if (err.code !== 11000) {
-        throw new Error('Failed to rename pages: ', err);
-      }
-    }
-
-    const newParentPath = path.replace(pathRegExp, newPagePathPrefix);
-    const newParentPage = await this.findByPath(newParentPath);
-    const renamedPages = await this.findManageableListWithDescendants(newParentPage, user, options);
-
-    pageEvent.emit('createMany', renamedPages, user, newParentPage);
-
-    // Execute after unorderedBulkOp to prevent duplication
-    if (createRedirectPage) {
-      await createRediectPageBulkOp.execute();
-    }
-
-    targetPage.path = newPagePathPrefix;
-    return targetPage;
-
-  };
-
   pageSchema.statics.findListByPathsArray = async function(paths) {
     const queryBuilder = new PageQueryBuilder(this.find());
     queryBuilder.addConditionToListByPathsArray(paths);

src/server/routes/apiv3/pages.js

@@ -4,7 +4,7 @@ const logger = loggerFactory('growi:routes:apiv3:pages'); // eslint-disable-line
 const express = require('express');
 const pathUtils = require('growi-commons').pathUtils;
 
-const { body } = require('express-validator/check');
+const { body } = require('express-validator');
 const { query } = require('express-validator');
 const ErrorV3 = require('../../models/vo/error-apiv3');
 
@@ -395,13 +395,7 @@ module.exports = (crowi) => {
       if (!page.isUpdatable(revisionId)) {
         return res.apiv3Err(new ErrorV3('Someone could update this page, so couldn\'t delete.', 'notfound_or_forbidden'), 409);
       }
-
-      if (isRecursively) {
-        page = await Page.renameRecursively(page, newPagePath, req.user, options);
-      }
-      else {
-        page = await Page.rename(page, newPagePath, req.user, options);
-      }
+      page = await crowi.pageService.renamePage(page, newPagePath, req.user, options, isRecursively);
     }
     catch (err) {
       logger.error(err);

src/server/routes/page.js

@@ -1343,12 +1343,7 @@ module.exports = function(crowi, app) {
         return res.json(ApiResponse.error('Someone could update this page, so couldn\'t delete.', 'outdated'));
       }
 
-      if (isRecursively) {
-        page = await Page.renameRecursively(page, newPagePath, req.user, options);
-      }
-      else {
-        page = await Page.rename(page, newPagePath, req.user, options);
-      }
+      page = await crowi.pageService.renamePage(page, newPagePath, req.user, options, isRecursively);
     }
     catch (err) {
       logger.error(err);

src/server/service/page.js

@@ -21,6 +21,140 @@ class PageService {
     this.pageEvent.on('createMany', this.pageEvent.onCreateMany);
   }
 
+
+  async renamePage(page, newPagePath, user, options, isRecursively = false) {
+
+    const Page = this.crowi.model('Page');
+    const Revision = this.crowi.model('Revision');
+    const path = page.path;
+    const createRedirectPage = options.createRedirectPage || false;
+    const updateMetadata = options.updateMetadata || false;
+    const socketClientId = options.socketClientId || null;
+
+    // sanitize path
+    newPagePath = this.crowi.xss.process(newPagePath); // eslint-disable-line no-param-reassign
+
+    const update = {};
+    // update Page
+    update.path = newPagePath;
+    if (updateMetadata) {
+      update.lastUpdateUser = user;
+      update.updatedAt = Date.now();
+    }
+    const renamedPage = await Page.findByIdAndUpdate(page._id, { $set: update }, { new: true });
+
+    // update Rivisions
+    await Revision.updateRevisionListByPath(path, { path: newPagePath }, {});
+
+    if (createRedirectPage) {
+      const body = `redirect ${newPagePath}`;
+      await Page.create(path, body, user, { redirectTo: newPagePath });
+    }
+
+    if (isRecursively) {
+      this.renameDescendantsWithStream(page, newPagePath, user, options);
+    }
+
+    this.pageEvent.emit('delete', page, user, socketClientId);
+    this.pageEvent.emit('create', renamedPage, user, socketClientId);
+
+    return renamedPage;
+  }
+
+
+  async renameDescendants(pages, user, options, oldPagePathPrefix, newPagePathPrefix) {
+    const Page = this.crowi.model('Page');
+
+    const pageCollection = mongoose.connection.collection('pages');
+    const revisionCollection = mongoose.connection.collection('revisions');
+    const { updateMetadata, createRedirectPage } = options;
+
+    const unorderedBulkOp = pageCollection.initializeUnorderedBulkOp();
+    const createRediectPageBulkOp = pageCollection.initializeUnorderedBulkOp();
+    const revisionUnorderedBulkOp = revisionCollection.initializeUnorderedBulkOp();
+
+    pages.forEach((page) => {
+      const newPagePath = page.path.replace(oldPagePathPrefix, newPagePathPrefix);
+      if (updateMetadata) {
+        unorderedBulkOp.find({ _id: page._id }).update([{ $set: { path: newPagePath, lastUpdateUser: user._id, updatedAt: { $toDate: Date.now() } } }]);
+      }
+      else {
+        unorderedBulkOp.find({ _id: page._id }).update({ $set: { path: newPagePath } });
+      }
+      if (createRedirectPage) {
+        createRediectPageBulkOp.insert({
+          path: page.path, body: `redirect ${newPagePath}`, creator: user, lastUpdateUser: user, status: Page.STATUS_PUBLISHED, redirectTo: newPagePath,
+        });
+      }
+      revisionUnorderedBulkOp.find({ path: page.path }).update({ $set: { path: newPagePath } }, { multi: true });
+    });
+
+    try {
+      await unorderedBulkOp.execute();
+      await revisionUnorderedBulkOp.execute();
+      // Execute after unorderedBulkOp to prevent duplication
+      if (createRedirectPage) {
+        await createRediectPageBulkOp.execute();
+      }
+    }
+    catch (err) {
+      if (err.code !== 11000) {
+        throw new Error('Failed to rename pages: ', err);
+      }
+    }
+
+    this.pageEvent.emit('updateMany', pages, user);
+  }
+
+  /**
+   * Create rename stream
+   */
+  async renameDescendantsWithStream(targetPage, newPagePath, user, options = {}) {
+    const Page = this.crowi.model('Page');
+    const newPagePathPrefix = newPagePath;
+    const { PageQueryBuilder } = Page;
+    const pathRegExp = new RegExp(`^${escapeStringRegexp(targetPage.path)}`, 'i');
+
+    const readStream = new PageQueryBuilder(Page.find())
+      .addConditionToExcludeRedirect()
+      .addConditionToListOnlyDescendants(targetPage.path)
+      .addConditionToFilteringByViewer(user)
+      .query
+      .lean()
+      .cursor();
+
+    const renameDescendants = this.renameDescendants.bind(this);
+    const pageEvent = this.pageEvent;
+    let count = 0;
+    const writeStream = new Writable({
+      objectMode: true,
+      async write(batch, encoding, callback) {
+        try {
+          count += batch.length;
+          await renameDescendants(batch, user, options, pathRegExp, newPagePathPrefix);
+          logger.debug(`Reverting pages progressing: (count=${count})`);
+        }
+        catch (err) {
+          logger.error('revertPages error on add anyway: ', err);
+        }
+
+        callback();
+      },
+      final(callback) {
+        logger.debug(`Reverting pages has completed: (totalCount=${count})`);
+        // update  path
+        targetPage.path = newPagePath;
+        pageEvent.emit('syncDescendants', targetPage, user);
+        callback();
+      },
+    });
+
+    readStream
+      .pipe(createBatchStream(BULK_REINDEX_SIZE))
+      .pipe(writeStream);
+  }
+
+
   async deleteCompletelyOperation(pageIds, pagePaths) {
     // Delete Bookmarks, Attachments, Revisions, Pages and emit delete
     const Bookmark = this.crowi.model('Bookmark');
@@ -180,6 +314,7 @@ class PageService {
       .cursor();
 
     const duplicateDescendants = this.duplicateDescendants.bind(this);
+    const pageEvent = this.pageEvent;
     let count = 0;
     const writeStream = new Writable({
       objectMode: true,
@@ -197,7 +332,9 @@ class PageService {
       },
       final(callback) {
         logger.debug(`Adding pages has completed: (totalCount=${count})`);
-
+        // update  path
+        page.path = newPagePath;
+        pageEvent.emit('syncDescendants', page, user);
         callback();
       },
     });
@@ -452,6 +589,8 @@ class PageService {
 
   async revertDeletedPage(page, user, options = {}, isRecursively = false) {
     const Page = this.crowi.model('Page');
+    const Revision = this.crowi.model('Revision');
+
     const newPath = Page.getRevertDeletedPageName(page.path);
     const originPage = await Page.findByPath(newPath);
     if (originPage != null) {
@@ -471,7 +610,11 @@ class PageService {
     page.status = Page.STATUS_PUBLISHED;
     page.lastUpdateUser = user;
     debug('Revert deleted the page', page, newPath);
-    const updatedPage = await Page.rename(page, newPath, user, {});
+    const updatedPage = await Page.findByIdAndUpdate(page._id, {
+      $set: { path: newPath, status: Page.STATUS_PUBLISHED, lastUpdateUser: user._id },
+    }, { new: true });
+    await Revision.updateMany({ path: page.path }, { $set: { path: newPath } });
+
     return updatedPage;
   }
 

src/server/service/search-delegator/elasticsearch.js

@@ -967,8 +967,8 @@ class ElasticsearchDelegator {
     return this.updateOrInsertPageById(page._id);
   }
 
-  async syncDescendantsPagesUpdated(pages, user, parentPage) {
-
+  // remove pages whitch should nod Indexed
+  async syncPagesUpdated(pages, user) {
     const shoudDeletePages = [];
     pages.forEach((page) => {
       logger.debug('SearchClient.syncPageUpdated', page.path);
@@ -986,7 +986,9 @@ class ElasticsearchDelegator {
     catch (err) {
       logger.error('deletePages:ES Error', err);
     }
+  }
 
+  async syncDescendantsPagesUpdated(parentPage, user) {
     return this.updateOrInsertDescendantsPagesById(parentPage, user);
   }
 

src/server/service/search.js

@@ -60,7 +60,8 @@ class SearchService {
     pageEvent.on('update', this.delegator.syncPageUpdated.bind(this.delegator));
     pageEvent.on('deleteCompletely', this.delegator.syncPagesDeletedCompletely.bind(this.delegator));
     pageEvent.on('delete', this.delegator.syncPageDeleted.bind(this.delegator));
-    pageEvent.on('createMany', this.delegator.syncDescendantsPagesUpdated.bind(this.delegator));
+    pageEvent.on('updateMany', this.delegator.syncPagesUpdated.bind(this.delegator));
+    pageEvent.on('syncDescendants', this.delegator.syncDescendantsPagesUpdated.bind(this.delegator));
 
     const bookmarkEvent = this.crowi.event('bookmark');
     bookmarkEvent.on('create', this.delegator.syncBookmarkChanged.bind(this.delegator));