Merge pull request #3300 from weseek/imprv/4685-4874-create-delete-stream

Imprv/4685 4874 create delete stream

src/server/routes/apiv3/pages.js

@@ -442,7 +442,7 @@ module.exports = (crowi) => {
     const options = { socketClientId };
 
     try {
-      const pages = await crowi.pageService.completelyDeletePageRecursively({ path: '/trash' }, req.user, options);
+      const pages = await crowi.pageService.deletePageRecursivelyCompletely({ path: '/trash' }, req.user, options);
       return res.apiv3({ pages });
     }
     catch (err) {

src/server/routes/page.js

@@ -1193,12 +1193,7 @@ module.exports = function(crowi, app) {
         if (!req.user.canDeleteCompletely(page.creator)) {
           return res.json(ApiResponse.error('You can not delete completely', 'user_not_admin'));
         }
-        if (isRecursively) {
-          await crowi.pageService.completelyDeletePageRecursively(page, req.user, options);
-        }
-        else {
-          await crowi.pageService.completelyDeleteSinglePage(page, req.user, options);
-        }
+        await crowi.pageService.deleteCompletely(page, req.user, options, isRecursively);
       }
       else {
         if (!page.isUpdatable(previousRevision)) {

src/server/service/page.js

@@ -15,7 +15,7 @@ class PageService {
     this.crowi = crowi;
   }
 
-  async deleteCompletely(pageIds, pagePaths) {
+  async deleteCompletelyOperation(pageIds, pagePaths) {
     // Delete Bookmarks, Attachments, Revisions, Pages and emit delete
     const Bookmark = this.crowi.model('Bookmark');
     const Comment = this.crowi.model('Comment');
@@ -164,7 +164,7 @@ class PageService {
   }
 
   // delete multiple pages
-  async completelyDeletePages(pagesData, user, options = {}) {
+  async deleteMultipleCompletely(pages, user, options = {}) {
     this.validateCrowi();
     let pageEvent;
     // init event
@@ -174,22 +174,21 @@ class PageService {
       pageEvent.on('update', pageEvent.onUpdate);
     }
 
-    const ids = pagesData.map(page => (page._id));
-    const paths = pagesData.map(page => (page.path));
+    const ids = pages.map(page => (page._id));
+    const paths = pages.map(page => (page.path));
     const socketClientId = options.socketClientId || null;
 
     logger.debug('Deleting completely', paths);
 
-    await this.deleteCompletely(ids, paths);
+    await this.deleteCompletelyOperation(ids, paths);
 
     if (socketClientId != null) {
-      pageEvent.emit('deleteCompletely', pagesData, user, socketClientId); // update as renamed page
+      pageEvent.emit('deleteCompletely', pages, user, socketClientId); // update as renamed page
     }
     return;
   }
 
-  // delete single page completely
-  async completelyDeleteSinglePage(pageData, user, options = {}) {
+  async deleteCompletely(page, user, options = {}, isRecursively = false) {
     this.validateCrowi();
     let pageEvent;
     // init event
@@ -199,32 +198,65 @@ class PageService {
       pageEvent.on('update', pageEvent.onUpdate);
     }
 
-    const ids = [pageData._id];
-    const paths = [pageData.path];
+    const ids = [page._id];
+    const paths = [page.path];
     const socketClientId = options.socketClientId || null;
 
     logger.debug('Deleting completely', paths);
 
-    await this.deleteCompletely(ids, paths);
+    await this.deleteCompletelyOperation(ids, paths);
+
+    if (isRecursively) {
+      this.deleteDescendantsWithStream(page, user, options);
+    }
 
     if (socketClientId != null) {
-      pageEvent.emit('delete', pageData, user, socketClientId); // update as renamed page
+      pageEvent.emit('delete', page, user, socketClientId); // update as renamed page
     }
     return;
   }
 
   /**
-   * Delete Bookmarks, Attachments, Revisions, Pages and emit delete
+   * Create delete stream
    */
-  async completelyDeletePageRecursively(targetPage, user, options = {}) {
-    const findOpts = { includeTrashed: true };
+  async deleteDescendantsWithStream(targetPage, user, options = {}) {
     const Page = this.crowi.model('Page');
+    const { PageQueryBuilder } = Page;
 
-    // find manageable descendants (this array does not include GRANT_RESTRICTED)
-    const pages = await Page.findManageableListWithDescendants(targetPage, user, findOpts);
+    const readStream = new PageQueryBuilder(Page.find())
+      .addConditionToExcludeRedirect()
+      .addConditionToListOnlyDescendants(targetPage.path)
+      .addConditionToFilteringByViewer(user)
+      .query
+      .lean()
+      .cursor();
+
+    const deleteMultipleCompletely = this.deleteMultipleCompletely.bind(this);
+    let count = 0;
+    const writeStream = new Writable({
+      objectMode: true,
+      async write(batch, encoding, callback) {
+        try {
+          count += batch.length;
+          await deleteMultipleCompletely(batch, user, options);
+          logger.debug(`Adding pages progressing: (count=${count})`);
+        }
+        catch (err) {
+          logger.error('addAllPages error on add anyway: ', err);
+        }
+
+        callback();
+      },
+      final(callback) {
+        logger.debug(`Adding pages has completed: (totalCount=${count})`);
 
-    // TODO streaming bellow action
-    return this.completelyDeletePages(pages, user, options);
+        callback();
+      },
+    });
+
+    readStream
+      .pipe(createBatchStream(BULK_REINDEX_SIZE))
+      .pipe(writeStream);
   }
 
   async revertDeletedPageRecursively(targetPage, user, options = {}) {
@@ -258,7 +290,7 @@ class PageService {
         throw new Error('The new page of to revert is exists and the redirect path of the page is not the deleted page.');
       }
       // originPage is object.
-      await this.completelyDeletePages([originPage], options);
+      await this.deleteMultiplePagesCompletely([originPage], options);
     }
 
     page.status = STATUS_PUBLISHED;
@@ -279,7 +311,7 @@ class PageService {
       if (originPage.redirectTo !== page.path) {
         throw new Error('The new page of to revert is exists and the redirect path of the page is not the deleted page.');
       }
-      await this.completelyDeleteSinglePage(originPage, options);
+      await this.deleteCompletely(originPage, options);
     }
 
     page.status = STATUS_PUBLISHED;
@@ -300,7 +332,7 @@ class PageService {
         }));
         break;
       case 'delete':
-        return this.completelyDeletePages(pages);
+        return this.deleteMultiplePagesCompletely(pages);
       case 'transfer':
         await Promise.all(pages.map((page) => {
           return Page.transferPageToGroup(page, transferToUserGroupId);