Handle broadcast in pipeline (#5732)

Signed-off-by: Denis Bykhov <bykhov.denis@gmail.com>

server/core/src/pipeline.ts

@@ -51,16 +51,27 @@ export async function createPipeline (
   upgrade: boolean,
   broadcast: BroadcastFunc
 ): Promise<Pipeline> {
+  const broadcastHandlers: BroadcastFunc[] = [broadcast]
+  const _broadcast: BroadcastFunc = (
+    tx: Tx[],
+    targets: string | string[] | undefined,
+    exclude: string[] | undefined
+  ) => {
+    for (const handler of broadcastHandlers) handler(tx, targets, exclude)
+  }
   const storage = await ctx.with(
     'create-server-storage',
     {},
     async (ctx) =>
       await createServerStorage(ctx, conf, {
         upgrade,
-        broadcast
+        broadcast: _broadcast
       })
   )
   const pipelineResult = await PipelineImpl.create(ctx.newChild('pipeline-operations', {}), storage, constructors)
+  broadcastHandlers.push((tx: Tx[], targets: string | string[] | undefined, exclude: string[] | undefined) => {
+    void pipelineResult.handleBroadcast(tx, targets, exclude)
+  })
   return pipelineResult
 }
 
@@ -115,6 +126,12 @@ class PipelineImpl implements Pipeline {
     }
   }
 
+  async handleBroadcast (tx: Tx[], targets?: string | string[], exclude?: string[]): Promise<void> {
+    if (this.head !== undefined) {
+      await this.head.handleBroadcast(tx, targets, exclude)
+    }
+  }
+
   async close (): Promise<void> {
     await this.storage.close()
   }

server/core/src/types.ts

@@ -94,6 +94,7 @@ export interface Middleware {
     query: DocumentQuery<T>,
     options?: FindOptions<T>
   ) => Promise<FindResult<T>>
+  handleBroadcast: HandleBroadcastFunc
   searchFulltext: (ctx: SessionContext, query: SearchQuery, options: SearchOptions) => Promise<SearchResult>
 }
 
@@ -101,10 +102,11 @@ export interface Middleware {
  * @public
  */
 export type BroadcastFunc = (tx: Tx[], targets?: string | string[], exclude?: string[]) => void
+
 /**
  * @public
  */
-export type HandledBroadcastFunc = (tx: Tx[], targets?: string[]) => Tx[]
+export type HandleBroadcastFunc = (tx: Tx[], targets?: string | string[], exclude?: string[]) => Promise<void>
 
 /**
  * @public

server/middleware/src/base.ts

@@ -49,6 +49,12 @@ export abstract class BaseMiddleware {
     return await this.provideSearchFulltext(ctx, query, options)
   }
 
+  async handleBroadcast (tx: Tx[], targets?: string | string[], exclude?: string[]): Promise<void> {
+    if (this.next !== undefined) {
+      await this.next.handleBroadcast(tx, targets, exclude)
+    }
+  }
+
   protected async provideTx (ctx: SessionContext, tx: Tx): Promise<TxMiddlewareResult> {
     if (this.next !== undefined) {
       return await this.next.tx(ctx, tx)

server/middleware/src/spaceSecurity.ts

@@ -88,6 +88,10 @@ export class SpaceSecurityMiddleware extends BaseMiddleware implements Middlewar
     return res
   }
 
+  private resyncDomains (): void {
+    this.spaceSecurityInit = this.init(this.spaceMeasureCtx)
+  }
+
   private addMemberSpace (member: Ref<Account>, space: Ref<Space>): void {
     const arr = this.allowedSpaces[member] ?? []
     arr.push(space)
@@ -393,6 +397,11 @@ export class SpaceSecurityMiddleware extends BaseMiddleware implements Middlewar
           await this.brodcastEvent(ctx, [cud.objectId])
         }
       }
+    } else if (tx._class === core.class.TxWorkspaceEvent) {
+      const event = tx as TxWorkspaceEvent
+      if (event.event === WorkspaceEvent.BulkUpdate) {
+        this.resyncDomains()
+      }
     }
   }
 
@@ -414,16 +423,27 @@ export class SpaceSecurityMiddleware extends BaseMiddleware implements Middlewar
       it.target = mergeTargets(targets, it.target)
     })
 
-    await this.waitInit()
-    for (const tt of ctx.derived) {
-      for (const t of tt.derived) {
-        if (this.storage.hierarchy.isDerived(t._class, core.class.TxCUD)) {
-          await this.processTxSpaceDomain(t as TxCUD<Doc>)
+    return res
+  }
+
+  override async handleBroadcast (
+    txes: Tx[],
+    targets?: string | string[] | undefined,
+    exclude?: string[] | undefined
+  ): Promise<void> {
+    for (const tx of txes) {
+      const h = this.storage.hierarchy
+      if (h.isDerived(tx._class, core.class.TxCUD)) {
+        const cudTx = tx as TxCUD<Doc>
+        await this.processTxSpaceDomain(cudTx)
+      } else if (tx._class === core.class.TxWorkspaceEvent) {
+        const event = tx as TxWorkspaceEvent
+        if (event.event === WorkspaceEvent.BulkUpdate) {
+          this.resyncDomains()
         }
       }
     }
-
-    return res
+    await this.next?.handleBroadcast(txes, targets, exclude)
   }
 
   private getAllAllowedSpaces (account: Account, isData: boolean): Ref<Space>[] {