Add workspace event send threshold and fix live query update

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

packages/presentation/src/utils.ts

@@ -152,6 +152,7 @@ export class LiveQuery {
   private oldQuery: DocumentQuery<Doc> | undefined
   private oldOptions: FindOptions<Doc> | undefined
   private oldCallback: ((result: FindResult<any>) => void) | undefined
+  private reqId = 0
   unsubscribe = () => {}
   clientRecreated = false
 
@@ -186,10 +187,16 @@ export class LiveQuery {
     callback: (result: FindResult<T>) => void,
     options: FindOptions<T> | undefined
   ): Promise<void> {
+    const id = ++this.reqId
     const piplineQuery = await pipeline.subscribe(_class, query, options, () => {
       // Refresh query if pipeline decide it is required.
       this.refreshClient()
     })
+    if (id !== this.reqId) {
+      // If we have one more request after this one, no need to do something.
+      piplineQuery.unsubscribe()
+      return
+    }
 
     this.unsubscribe()
     this.oldCallback = callback

server/core/src/indexer/indexer.ts

@@ -252,6 +252,9 @@ export class FullTextIndexPipeline implements FullTextPipeline {
     }
   }
 
+  broadcastClasses = new Set<Ref<Class<Doc>>>()
+  updateBroadcast: any = undefined
+
   async doIndexing (): Promise<void> {
     // Check model is upgraded to support indexer.
 
@@ -262,23 +265,25 @@ export class FullTextIndexPipeline implements FullTextPipeline {
       return
     }
     await this.initStates()
-    const classes = new Set<Ref<Class<Doc>>>()
     while (!this.cancelling) {
       await this.initializeStages()
       await this.processRemove()
 
       console.log('Indexing:', this.indexId, this.workspace)
       const _classes = await rateLimitter.exec(() => this.processIndex())
-      _classes.forEach((it) => classes.add(it))
+      _classes.forEach((it) => this.broadcastClasses.add(it))
 
       if (this.toIndex.size === 0 || this.stageChanged === 0) {
         if (this.toIndex.size === 0) {
           console.log(`${this.workspace.name} Indexing complete`, this.indexId)
         }
         if (!this.cancelling) {
           // We need to send index update event
-          this.broadcastUpdate(Array.from(classes.values()))
-          classes.clear()
+          clearTimeout(this.updateBroadcast)
+          this.updateBroadcast = setTimeout(() => {
+            this.broadcastUpdate(Array.from(this.broadcastClasses.values()))
+            this.broadcastClasses.clear()
+          }, 5000)
 
           await new Promise((resolve) => {
             this.triggerIndexing = () => {

server/core/src/storage.ts

@@ -816,6 +816,7 @@ export async function createServerStorage (
       throw new Error('No storage adapter')
     }
     const stages = conf.fulltextAdapter.stages(fulltextAdapter, storage, storageAdapter, contentAdapter)
+
     const indexer = new FullTextIndexPipeline(
       defaultAdapter,
       stages,