Merge pull request #6077 from weseek/imprv/execute-socket-emit-only-if-reindex

imprv: Add shouldEmit argument and excute socket.emit only in addAllPages

packages/app/src/server/interfaces/search.ts

@@ -36,6 +36,11 @@ export type SearchableData<T = Partial<QueryTerms>> = {
   terms: T
 }
 
+export type UpdateOrInsertPagesOpts = {
+  shouldEmitProgress?: boolean
+  invokeGarbageCollection?: boolean
+}
+
 // Terms Key types
 export type AllTermsKey = keyof QueryTerms;
 export type UnavailableTermsKey<K extends AllTermsKey> = Exclude<AllTermsKey, K>;

packages/app/src/server/interfaces/page-operation.ts → packages/app/src/server/models/interfaces/page-operation.ts

@@ -1,4 +1,4 @@
-import { ObjectIdLike } from './mongoose-utils';
+import { ObjectIdLike } from '../../interfaces/mongoose-utils';
 
 export type IPageForResuming = {
   _id: ObjectIdLike,

packages/app/src/server/models/page-operation.ts

@@ -6,7 +6,7 @@ import mongoose, {
 
 import {
   IPageForResuming, IUserForResuming, IOptionsForResuming,
-} from '~/server/interfaces/page-operation';
+} from '~/server/models/interfaces/page-operation';
 
 import loggerFactory from '../../utils/logger';
 import { ObjectIdLike } from '../interfaces/mongoose-utils';

packages/app/src/server/service/interfaces/search.ts

@@ -0,0 +1,4 @@
+export type UpdateOrInsertPagesOpts = {
+  shouldEmitProgress?: boolean
+  invokeGarbageCollection?: boolean
+}

packages/app/src/server/service/page.ts

@@ -2713,7 +2713,7 @@ class PageService {
 
     // then migrate
     try {
-      await this.normalizeParentRecursively(['/'], null);
+      await this.normalizeParentRecursively(['/'], null, true);
     }
     catch (err) {
       logger.error('V5 initial miration failed.', err);
@@ -2760,7 +2760,7 @@ class PageService {
    * @param user To be used to filter pages to update. If null, only public pages will be updated.
    * @returns Promise<void>
    */
-  async normalizeParentRecursively(paths: string[], user: any | null, shouldEmit = false): Promise<number> {
+  async normalizeParentRecursively(paths: string[], user: any | null, shouldEmitProgress = false): Promise<number> {
     const Page = mongoose.model('Page') as unknown as PageModel;
 
     const ancestorPaths = paths.flatMap(p => collectAncestorPaths(p, []));
@@ -2779,7 +2779,7 @@ class PageService {
 
     const grantFiltersByUser: { $or: any[] } = Page.generateGrantCondition(user, userGroups);
 
-    return this._normalizeParentRecursively(pathAndRegExpsToNormalize, ancestorPaths, grantFiltersByUser, user, shouldEmit);
+    return this._normalizeParentRecursively(pathAndRegExpsToNormalize, ancestorPaths, grantFiltersByUser, user, shouldEmitProgress);
   }
 
   private buildFilterForNormalizeParentRecursively(pathOrRegExps: (RegExp | string)[], publicPathsToNormalize: string[], grantFiltersByUser: { $or: any[] }) {
@@ -2829,7 +2829,7 @@ class PageService {
       publicPathsToNormalize: string[],
       grantFiltersByUser: { $or: any[] },
       user,
-      shouldEmit = false,
+      shouldEmitProgress = false,
       count = 0,
       skiped = 0,
       isFirst = true,
@@ -2837,7 +2837,7 @@ class PageService {
     const BATCH_SIZE = 100;
     const PAGES_LIMIT = 1000;
 
-    const socket = shouldEmit ? this.crowi.socketIoService.getAdminSocket() : null;
+    const socket = shouldEmitProgress ? this.crowi.socketIoService.getAdminSocket() : null;
 
     const Page = mongoose.model('Page') as unknown as PageModel;
     const { PageQueryBuilder } = Page;
@@ -3002,7 +3002,16 @@ class PageService {
     await streamToPromise(migratePagesStream);
 
     if (await Page.exists(matchFilter) && shouldContinue) {
-      return this._normalizeParentRecursively(pathOrRegExps, publicPathsToNormalize, grantFiltersByUser, user, shouldEmit, nextCount, nextSkiped, false);
+      return this._normalizeParentRecursively(
+        pathOrRegExps,
+        publicPathsToNormalize,
+        grantFiltersByUser,
+        user,
+        shouldEmitProgress,
+        nextCount,
+        nextSkiped,
+        false,
+      );
     }
 
     // End

packages/app/src/server/service/search-delegator/elasticsearch.ts

@@ -18,6 +18,7 @@ import {
 } from '../../interfaces/search';
 import { PageModel } from '../../models/page';
 import { createBatchStream } from '../../util/batch-stream';
+import { UpdateOrInsertPagesOpts } from '../interfaces/search';
 
 
 import ElasticsearchClient from './elasticsearch-client';
@@ -437,7 +438,7 @@ class ElasticsearchDelegator implements SearchDelegator<Data, ESTermsKey, ESQuer
 
   addAllPages() {
     const Page = mongoose.model('Page');
-    return this.updateOrInsertPages(() => Page.find(), { isEmittingProgressEvent: true, invokeGarbageCollection: true });
+    return this.updateOrInsertPages(() => Page.find(), { shouldEmitProgress: true, invokeGarbageCollection: true });
   }
 
   updateOrInsertPageById(pageId) {
@@ -456,8 +457,8 @@ class ElasticsearchDelegator implements SearchDelegator<Data, ESTermsKey, ESQuer
   /**
    * @param {function} queryFactory factory method to generate a Mongoose Query instance
    */
-  async updateOrInsertPages(queryFactory, option: any = {}) {
-    const { isEmittingProgressEvent = false, invokeGarbageCollection = false } = option;
+  async updateOrInsertPages(queryFactory, option: UpdateOrInsertPagesOpts = {}) {
+    const { shouldEmitProgress = false, invokeGarbageCollection = false } = option;
 
     const Page = mongoose.model('Page') as unknown as PageModel;
     const { PageQueryBuilder } = Page;
@@ -465,7 +466,7 @@ class ElasticsearchDelegator implements SearchDelegator<Data, ESTermsKey, ESQuer
     const Comment = mongoose.model('Comment') as any; // TODO: typescriptize model
     const PageTagRelation = mongoose.model('PageTagRelation') as any; // TODO: typescriptize model
 
-    const socket = this.socketIoService.getAdminSocket();
+    const socket = shouldEmitProgress ? this.socketIoService.getAdminSocket() : null;
 
     // prepare functions invoked from custom streams
     const prepareBodyForCreate = this.prepareBodyForCreate.bind(this);
@@ -583,8 +584,8 @@ class ElasticsearchDelegator implements SearchDelegator<Data, ESTermsKey, ESQuer
 
           logger.info(`Adding pages progressing: (count=${count}, errors=${res.errors}, took=${res.took}ms)`);
 
-          if (isEmittingProgressEvent) {
-            socket.emit('addPageProgress', { totalCount, count, skipped });
+          if (shouldEmitProgress) {
+            socket?.emit('addPageProgress', { totalCount, count, skipped });
           }
         }
         catch (err) {
@@ -607,8 +608,8 @@ class ElasticsearchDelegator implements SearchDelegator<Data, ESTermsKey, ESQuer
       final(callback) {
         logger.info(`Adding pages has completed: (totalCount=${totalCount}, skipped=${skipped})`);
 
-        if (isEmittingProgressEvent) {
-          socket.emit('finishAddPage', { totalCount, count, skipped });
+        if (shouldEmitProgress) {
+          socket?.emit('finishAddPage', { totalCount, count, skipped });
         }
         callback();
       },
@@ -623,7 +624,6 @@ class ElasticsearchDelegator implements SearchDelegator<Data, ESTermsKey, ESQuer
       .pipe(writeStream);
 
     return streamToPromise(writeStream);
-
   }
 
   deletePages(pages) {