UBERF-4319: Fix performance issues

Signed-off-by: Andrey Sobolev <haiodo@gmail.com>

models/core/src/index.ts

@@ -204,4 +204,8 @@ export function createModel (builder: Builder): void {
       ]
     }
   )
+
+  builder.mixin(core.class.Space, core.class.Class, core.mixin.FullTextSearchContext, {
+    childProcessingAllowed: false
+  })
 }

models/recruit/src/index.ts

@@ -1655,12 +1655,14 @@ export function createModel (builder: Builder): void {
   // Allow to use fuzzy search for mixins
   builder.mixin(recruit.class.Vacancy, core.class.Class, core.mixin.FullTextSearchContext, {
     fullTextSummary: true,
+    childProcessingAllowed: true,
     propagate: []
   })
 
   builder.mixin(recruit.mixin.Candidate, core.class.Class, core.mixin.FullTextSearchContext, {
     fullTextSummary: true,
     propagate: [recruit.class.Applicant],
+    childProcessingAllowed: true,
     propagateClasses: [
       tags.class.TagReference,
       chunter.class.ChatMessage,
@@ -1673,6 +1675,7 @@ export function createModel (builder: Builder): void {
   builder.mixin(recruit.class.Applicant, core.class.Class, core.mixin.FullTextSearchContext, {
     fullTextSummary: true,
     forceIndex: true,
+    childProcessingAllowed: true,
     propagate: []
   })
 

models/telegram/src/index.ts

@@ -251,6 +251,7 @@ export function createModel (builder: Builder): void {
   )
 
   builder.mixin(telegram.class.Message, core.class.Class, core.mixin.FullTextSearchContext, {
-    parentPropagate: false
+    parentPropagate: false,
+    childProcessingAllowed: true
   })
 }

packages/core/src/classes.ts

@@ -471,6 +471,8 @@ export interface FullTextSearchContext extends Class<Doc> {
 
   // Do we need to propagate child value to parent one. Default(true)
   parentPropagate?: boolean
+
+  childProcessingAllowed?: boolean
 }
 
 /**

packages/query/src/index.ts

@@ -54,6 +54,7 @@ import core, {
   resultSort,
   toFindResult
 } from '@hcengineering/core'
+import { PlatformError } from '@hcengineering/platform'
 import { deepEqual } from 'fast-equals'
 
 const CACHE_SIZE = 100
@@ -105,7 +106,12 @@ export class LiveQuery extends TxProcessor implements Client {
       if (!this.removeFromQueue(q)) {
         try {
           await this.refresh(q)
-        } catch (err) {
+        } catch (err: any) {
+          if (err instanceof PlatformError) {
+            if (err.message === 'connection closed') {
+              continue
+            }
+          }
           console.error(err)
         }
       }
@@ -114,7 +120,12 @@ export class LiveQuery extends TxProcessor implements Client {
       for (const q of v) {
         try {
           await this.refresh(q)
-        } catch (err) {
+        } catch (err: any) {
+          if (err instanceof PlatformError) {
+            if (err.message === 'connection closed') {
+              continue
+            }
+          }
           console.error(err)
         }
       }

server/core/src/fulltext.ts

@@ -19,33 +19,33 @@ import core, {
   Class,
   Doc,
   DocIndexState,
-  docKey,
   DocumentQuery,
   FindOptions,
   FindResult,
   Hierarchy,
-  isFullTextAttribute,
-  isIndexedAttribute,
   MeasureContext,
   ObjQueryType,
   Ref,
+  SearchOptions,
+  SearchQuery,
+  SearchResult,
   ServerStorage,
-  toFindResult,
   Tx,
-  TxCollectionCUD,
   TxCUD,
+  TxCollectionCUD,
   TxFactory,
   TxResult,
   WorkspaceId,
-  SearchQuery,
-  SearchOptions,
-  SearchResult
+  docKey,
+  isFullTextAttribute,
+  isIndexedAttribute,
+  toFindResult
 } from '@hcengineering/core'
 import { MinioService } from '@hcengineering/minio'
 import { FullTextIndexPipeline } from './indexer'
 import { createStateDoc, isClassIndexable } from './indexer/utils'
-import { mapSearchResultDoc, getScoringConfig } from './mapper'
-import type { FullTextAdapter, WithFind, IndexedDoc } from './types'
+import { getScoringConfig, mapSearchResultDoc } from './mapper'
+import type { FullTextAdapter, IndexedDoc, WithFind } from './types'
 
 /**
  * @public
@@ -79,44 +79,58 @@ export class FullTextIndex implements WithFind {
     await this.consistency
   }
 
-  async tx (ctx: MeasureContext, tx: Tx): Promise<TxResult> {
-    let attachedTo: Ref<DocIndexState> | undefined
-    let attachedToClass: Ref<Class<Doc>> | undefined
-    if (tx._class === core.class.TxCollectionCUD) {
-      const txcol = tx as TxCollectionCUD<Doc, AttachedDoc>
-      attachedTo = txcol.objectId as Ref<DocIndexState>
-      attachedToClass = txcol.objectClass
-      tx = txcol.tx
-    }
-    if (this.hierarchy.isDerived(tx._class, core.class.TxCUD)) {
-      const cud = tx as TxCUD<Doc>
-
-      if (!isClassIndexable(this.hierarchy, cud.objectClass)) {
-        // No need, since no indixable fields or attachments.
-        return {}
+  async tx (ctx: MeasureContext, txes: Tx[]): Promise<TxResult> {
+    const stDocs = new Map<Ref<DocIndexState>, { create?: DocIndexState, updated: boolean, removed: boolean }>()
+    for (let tx of txes) {
+      let attachedTo: Ref<DocIndexState> | undefined
+      let attachedToClass: Ref<Class<Doc>> | undefined
+      if (tx._class === core.class.TxCollectionCUD) {
+        const txcol = tx as TxCollectionCUD<Doc, AttachedDoc>
+        attachedTo = txcol.objectId as Ref<DocIndexState>
+        attachedToClass = txcol.objectClass
+        tx = txcol.tx
       }
+      if (this.hierarchy.isDerived(tx._class, core.class.TxCUD)) {
+        const cud = tx as TxCUD<Doc>
 
-      let stDoc: DocIndexState | undefined
-      if (cud._class === core.class.TxCreateDoc) {
-        // Add doc for indexing
-        stDoc = createStateDoc(cud.objectId, cud.objectClass, {
-          attributes: {},
-          stages: {},
-          attachedTo,
-          attachedToClass,
-          space: tx.objectSpace,
-          removed: false
-        })
-      }
-      await this.indexer.queue(
-        cud.objectId as Ref<DocIndexState>,
-        cud._class === core.class.TxCreateDoc,
-        cud._class === core.class.TxRemoveDoc,
-        stDoc
-      )
+        if (!isClassIndexable(this.hierarchy, cud.objectClass)) {
+          // No need, since no indixable fields or attachments.
+          continue
+        }
 
-      this.indexer.triggerIndexing()
+        let stDoc: DocIndexState | undefined
+        if (cud._class === core.class.TxCreateDoc) {
+          // Add doc for indexing
+          stDoc = createStateDoc(cud.objectId, cud.objectClass, {
+            attributes: {},
+            stages: {},
+            attachedTo,
+            attachedToClass,
+            space: tx.objectSpace,
+            removed: false
+          })
+          stDocs.set(cud.objectId as Ref<DocIndexState>, { create: stDoc, updated: false, removed: false })
+        } else {
+          const old = stDocs.get(cud.objectId as Ref<DocIndexState>)
+          if (cud._class === core.class.TxRemoveDoc && old?.create !== undefined) {
+            // Object created and deleted, skip index
+            continue
+          } else if (old !== undefined) {
+            // Create and update
+            // Skip update
+            continue
+          }
+          stDocs.set(cud.objectId as Ref<DocIndexState>, {
+            updated: cud._class !== core.class.TxRemoveDoc,
+            removed: cud._class === core.class.TxRemoveDoc
+          })
+        }
+      }
     }
+    await ctx.with('queue', {}, async (ctx) => {
+      await this.indexer.queue(ctx, stDocs)
+    })
+    this.indexer.triggerIndexing()
     return {}
   }
 

server/core/src/indexer/field.ts

@@ -117,9 +117,15 @@ export class IndexedFieldStage implements FullTextPipelineStage {
       // Obtain real documents
       const valueIds = new Map(values.map((it) => [it._id, it]))
       const objClass = v as Ref<Class<Doc>>
-      const docs = await this.dbStorage.findAll(metrics, objClass, {
-        _id: { $in: Array.from(valueIds.keys()) }
-      })
+      const kids = Array.from(valueIds.keys())
+      const docs = await this.dbStorage.findAll(
+        metrics,
+        objClass,
+        {
+          _id: kids.length === 1 ? kids[0] : { $in: kids }
+        },
+        { limit: kids.length }
+      )
       const attributes = getFullTextIndexableAttributes(pipeline.hierarchy, objClass)
 
       // Child docs.
@@ -197,11 +203,21 @@ export class IndexedFieldStage implements FullTextPipelineStage {
           if (propagate.length > 0) {
             // We need to propagate all changes to all child's of following classes.
             if (allChildDocs === undefined) {
-              const pc = metrics.newChild('propagate', {})
-              allChildDocs = await this.dbStorage.findAll(pc, core.class.DocIndexState, {
-                attachedTo: { $in: docs.map((it) => it._id) }
-              })
-              pc.end()
+              const ids = docs.map((it) => it._id)
+
+              allChildDocs = await metrics.with(
+                'propagate',
+                {},
+                async (ctx) =>
+                  await this.dbStorage.findAll(
+                    ctx,
+                    core.class.DocIndexState,
+                    {
+                      attachedTo: ids.length === 1 ? ids[0] : { $in: ids }
+                    },
+                    { limit: ids.length }
+                  )
+              )
             }
             const childs = allChildDocs.filter((it) => it.attachedTo === docState._id)
             for (const u of childs) {