TSK-656 (#2636)

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

server/core/src/pipeline.ts

@@ -36,10 +36,12 @@ import { Middleware, MiddlewareCreator, Pipeline, SessionContext } from './types
 export async function createPipeline (
   conf: DbConfiguration,
   constructors: MiddlewareCreator[],
-  upgrade: boolean
+  upgrade: boolean,
+  broadcast: (tx: Tx[]) => void
 ): Promise<Pipeline> {
   const storage = await createServerStorage(conf, {
-    upgrade
+    upgrade,
+    broadcast
   })
   return new TPipeline(storage, constructors)
 }

server/core/src/plugin.ts

@@ -18,7 +18,7 @@ import type { Plugin } from '@hcengineering/platform'
 import { plugin } from '@hcengineering/platform'
 
 import type { Class, Ref, Space } from '@hcengineering/core'
-import type { ObjectDDParticipant, Trigger } from './types'
+import type { AsyncTrigger, AsyncTriggerState, ObjectDDParticipant, Trigger } from './types'
 
 /**
  * @public
@@ -30,13 +30,16 @@ export const serverCoreId = 'server-core' as Plugin
  */
 const serverCore = plugin(serverCoreId, {
   class: {
-    Trigger: '' as Ref<Class<Trigger>>
+    Trigger: '' as Ref<Class<Trigger>>,
+    AsyncTrigger: '' as Ref<Class<AsyncTrigger>>,
+    AsyncTriggerState: '' as Ref<Class<AsyncTriggerState>>
   },
   mixin: {
     ObjectDDParticipant: '' as Ref<ObjectDDParticipant>
   },
   space: {
-    DocIndexState: '' as Ref<Space>
+    DocIndexState: '' as Ref<Space>,
+    TriggerState: '' as Ref<Space>
   }
 })
 

server/core/src/processor/index.ts

@@ -0,0 +1,123 @@
+import core, {
+  Class,
+  Doc,
+  Hierarchy,
+  MeasureContext,
+  ModelDb,
+  Ref,
+  ServerStorage,
+  Tx,
+  TxCUD,
+  TxFactory,
+  TxProcessor
+} from '@hcengineering/core'
+import { getResource } from '@hcengineering/platform'
+import plugin from '../plugin'
+import { AsyncTrigger, AsyncTriggerControl, AsyncTriggerFunc } from '../types'
+/**
+ * @public
+ */
+export class AsyncTriggerProcessor {
+  canceling: boolean = false
+
+  processing: Promise<void> | undefined
+
+  triggers: AsyncTrigger[] = []
+
+  classes: Ref<Class<Doc>>[] = []
+
+  factory = new TxFactory(core.account.System)
+
+  functions: AsyncTriggerFunc[] = []
+
+  trigger = (): void => {}
+
+  control: AsyncTriggerControl
+
+  constructor (
+    readonly model: ModelDb,
+    readonly hierarchy: Hierarchy,
+    readonly storage: ServerStorage,
+    readonly metrics: MeasureContext
+  ) {
+    this.control = {
+      hierarchy: this.hierarchy,
+      modelDb: this.model,
+      txFactory: this.factory,
+      findAll: async (_class, query, options) => {
+        return await this.storage.findAll(this.metrics, _class, query, options)
+      }
+    }
+  }
+
+  async cancel (): Promise<void> {
+    this.canceling = true
+    await this.processing
+  }
+
+  async start (): Promise<void> {
+    await this.updateTriggers()
+    this.processing = this.doProcessing()
+  }
+
+  async updateTriggers (): Promise<void> {
+    try {
+      this.triggers = await this.model.findAll(plugin.class.AsyncTrigger, {})
+      this.classes = this.triggers.reduce<Ref<Class<Doc>>[]>((arr, it) => arr.concat(it.classes), [])
+      this.functions = await Promise.all(this.triggers.map(async (trigger) => await getResource(trigger.trigger)))
+    } catch (err: any) {
+      console.error(err)
+    }
+  }
+
+  async tx (tx: Tx[]): Promise<void> {
+    const result: Tx[] = []
+    for (const _tx of tx) {
+      const actualTx = TxProcessor.extractTx(_tx)
+      if (
+        this.hierarchy.isDerived(actualTx._class, core.class.TxCUD) &&
+        this.hierarchy.isDerived(_tx._class, core.class.TxCUD)
+      ) {
+        const cud = actualTx as TxCUD<Doc>
+        if (this.classes.some((it) => this.hierarchy.isDerived(cud.objectClass, it))) {
+          // We need processing
+          result.push(
+            this.factory.createTxCreateDoc(plugin.class.AsyncTriggerState, plugin.space.TriggerState, {
+              tx: _tx as TxCUD<Doc>,
+              message: 'Processing...'
+            })
+          )
+        }
+      }
+    }
+    if (result.length > 0) {
+      await this.storage.apply(this.metrics, result, false)
+      this.processing = this.doProcessing()
+    }
+  }
+
+  private async doProcessing (): Promise<void> {
+    while (!this.canceling) {
+      const docs = await this.storage.findAll(this.metrics, plugin.class.AsyncTriggerState, {}, { limit: 10 })
+      if (docs.length === 0) {
+        return
+      }
+
+      for (const doc of docs) {
+        const result: Tx[] = []
+        if (this.canceling) {
+          break
+        }
+
+        try {
+          for (const f of this.functions) {
+            result.push(...(await f(doc.tx, this.control)))
+          }
+        } catch (err: any) {}
+        await this.storage.apply(this.metrics, [this.factory.createTxRemoveDoc(doc._class, doc.space, doc._id)], false)
+
+        await this.storage.apply(this.metrics, result, true)
+      }
+    }
+  }
+}

server/core/src/storage.ts

@@ -54,13 +54,15 @@ import { FullTextIndex } from './fulltext'
 import { FullTextIndexPipeline } from './indexer'
 import { FullTextPipelineStage } from './indexer/types'
 import serverCore from './plugin'
+import { AsyncTriggerProcessor } from './processor'
 import { Triggers } from './triggers'
 import type {
   ContentAdapterFactory,
   ContentTextAdapter,
   FullTextAdapter,
   FullTextAdapterFactory,
-  ObjectDDParticipant
+  ObjectDDParticipant,
+  TriggerControl
 } from './types'
 import { createCacheFindAll } from './utils'
 
@@ -82,16 +84,15 @@ export interface DbConfiguration {
   domains: Record<string, string>
   defaultAdapter: string
   workspace: WorkspaceId
+  metrics: MeasureContext
   fulltextAdapter: {
     factory: FullTextAdapterFactory
     url: string
-    metrics: MeasureContext
     stages: FullTextPipelineStageFactory
   }
   contentAdapter: {
     factory: ContentAdapterFactory
     url: string
-    metrics: MeasureContext
   }
   storageFactory?: () => MinioService
 }
@@ -99,6 +100,7 @@ export interface DbConfiguration {
 class TServerStorage implements ServerStorage {
   private readonly fulltext: FullTextIndex
   hierarchy: Hierarchy
+  triggerProcessor: AsyncTriggerProcessor
 
   scopes = new Map<string, Promise<any>>()
 
@@ -112,16 +114,19 @@ class TServerStorage implements ServerStorage {
     readonly storageAdapter: MinioService | undefined,
     readonly modelDb: ModelDb,
     private readonly workspace: WorkspaceId,
-    private readonly contentAdapter: ContentTextAdapter,
     readonly indexFactory: (storage: ServerStorage) => FullTextIndex,
-    options?: ServerStorageOptions
+    readonly options: ServerStorageOptions,
+    metrics: MeasureContext
   ) {
     this.hierarchy = hierarchy
     this.fulltext = indexFactory(this)
+    this.triggerProcessor = new AsyncTriggerProcessor(modelDb, hierarchy, this, metrics.newChild('triggers', {}))
+    void this.triggerProcessor.start()
   }
 
   async close (): Promise<void> {
     await this.fulltext.close()
+    await this.triggerProcessor.cancel()
     for (const o of this.adapters.values()) {
       await o.close()
     }
@@ -549,31 +554,28 @@ class TServerStorage implements ServerStorage {
     )
     const moves = await ctx.with('process-move', {}, () => this.processMove(ctx, txes, findAll))
 
+    const triggerControl: Omit<TriggerControl, 'txFactory'> = {
+      removedMap,
+      workspace: this.workspace,
+      fx: triggerFx.fx,
+      fulltextFx: (f) => triggerFx.fx(() => f(this.fulltextAdapter)),
+      storageFx: (f) => {
+        const adapter = this.storageAdapter
+        if (adapter === undefined) {
+          return
+        }
+
+        triggerFx.fx(() => f(adapter, this.workspace))
+      },
+      findAll: fAll(ctx),
+      modelDb: this.modelDb,
+      hierarchy: this.hierarchy
+    }
     const triggers = await ctx.with('process-triggers', {}, async (ctx) => {
       const result: Tx[] = []
       for (const tx of txes) {
-        result.push(
-          ...(await this.triggers.apply(tx.modifiedBy, tx, {
-            removedMap,
-            workspace: this.workspace,
-            fx: triggerFx.fx,
-            fulltextFx: (f) => triggerFx.fx(() => f(this.fulltextAdapter)),
-            storageFx: (f) => {
-              const adapter = this.storageAdapter
-              if (adapter === undefined) {
-                return
-              }
-
-              triggerFx.fx(() => f(adapter, this.workspace))
-            },
-            findAll: fAll(ctx),
-            modelDb: this.modelDb,
-            hierarchy: this.hierarchy,
-            txFx: async (f) => {
-              await f(this.getAdapter(DOMAIN_TX))
-            }
-          }))
-        )
+        result.push(...(await this.triggers.apply(tx.modifiedBy, tx, triggerControl)))
+        await ctx.with('async-triggers', {}, (ctx) => this.triggerProcessor.tx([tx]))
       }
       return result
     })
@@ -639,6 +641,40 @@ class TServerStorage implements ServerStorage {
     return { passed, onEnd }
   }
 
+  async apply (ctx: MeasureContext, tx: Tx[], broadcast: boolean): Promise<Tx[]> {
+    const triggerFx = new Effects()
+    const cacheFind = createCacheFindAll(this)
+
+    const txToStore = tx.filter(
+      (it) => it.space !== core.space.DerivedTx && !this.hierarchy.isDerived(it._class, core.class.TxApplyIf)
+    )
+    await ctx.with('domain-tx', {}, async () => await this.getAdapter(DOMAIN_TX).tx(...txToStore))
+
+    await ctx.with('apply', {}, (ctx) => this.routeTx(ctx, ...tx))
+
+    // send transactions
+    if (broadcast) {
+      this.options?.broadcast?.(tx)
+    }
+    // invoke triggers and store derived objects
+    const derived = await this.proccessDerived(ctx, tx, triggerFx, cacheFind, new Map<Ref<Doc>, Doc>())
+
+    // index object
+    for (const _tx of tx) {
+      await ctx.with('fulltext', {}, (ctx) => this.fulltext.tx(ctx, _tx))
+    }
+
+    // index derived objects
+    for (const tx of derived) {
+      await ctx.with('derived-processor', { _class: txClass(tx) }, (ctx) => this.fulltext.tx(ctx, tx))
+    }
+
+    for (const fx of triggerFx.effects) {
+      await fx()
+    }
+    return [...tx, ...derived]
+  }
+
   async tx (ctx: MeasureContext, tx: Tx): Promise<[TxResult, Tx[]]> {
     // store tx
     const _class = txClass(tx)
@@ -753,13 +789,15 @@ export interface ServerStorageOptions {
 
   // Indexing is not required to be started for upgrade mode.
   upgrade: boolean
+
+  broadcast?: (tx: Tx[]) => void
 }
 /**
  * @public
  */
 export async function createServerStorage (
   conf: DbConfiguration,
-  options?: ServerStorageOptions
+  options: ServerStorageOptions
 ): Promise<ServerStorage> {
   const hierarchy = new Hierarchy()
   const triggers = new Triggers()
@@ -803,13 +841,15 @@ export async function createServerStorage (
   const fulltextAdapter = await conf.fulltextAdapter.factory(
     conf.fulltextAdapter.url,
     conf.workspace,
-    conf.fulltextAdapter.metrics
+    conf.metrics.newChild('fulltext', {})
   )
 
+  const metrics = conf.metrics.newChild('server-storage', {})
+
   const contentAdapter = await conf.contentAdapter.factory(
     conf.contentAdapter.url,
     conf.workspace,
-    conf.contentAdapter.metrics
+    metrics.newChild('content', {})
   )
 
   const defaultAdapter = adapters.get(conf.defaultAdapter)
@@ -827,7 +867,7 @@ export async function createServerStorage (
       stages,
       hierarchy,
       conf.workspace,
-      fulltextAdapter.metrics(),
+      metrics.newChild('fulltext', {}),
       modelDb
     )
     return new FullTextIndex(
@@ -837,10 +877,9 @@ export async function createServerStorage (
       storageAdapter,
       conf.workspace,
       indexer,
-      options?.upgrade ?? false
+      options.upgrade ?? false
     )
   }
-
   return new TServerStorage(
     conf.domains,
     conf.defaultAdapter,
@@ -851,9 +890,9 @@ export async function createServerStorage (
     storageAdapter,
     modelDb,
     conf.workspace,
-    contentAdapter,
     indexFactory,
-    options
+    options,
+    metrics
   )
 }
 

server/core/src/types.ts

@@ -32,6 +32,7 @@ import {
   Storage,
   Timestamp,
   Tx,
+  TxCUD,
   TxFactory,
   TxResult,
   WorkspaceId
@@ -112,8 +113,6 @@ export interface TriggerControl {
   // Later can be replaced with generic one with bucket encapsulated inside.
   storageFx: (f: (adapter: MinioService, workspaceId: WorkspaceId) => Promise<void>) => void
   fx: (f: () => Promise<void>) => void
-
-  txFx: (f: (storage: Storage) => Promise<void>) => Promise<void>
 }
 
 /**
@@ -121,6 +120,20 @@ export interface TriggerControl {
  */
 export type TriggerFunc = (tx: Tx, ctrl: TriggerControl) => Promise<Tx[]>
 
+/**
+ * @public
+ */
+export interface AsyncTriggerControl {
+  txFactory: TxFactory
+  findAll: Storage['findAll']
+  hierarchy: Hierarchy
+  modelDb: ModelDb
+}
+/**
+ * @public
+ */
+export type AsyncTriggerFunc = (tx: Tx, ctrl: AsyncTriggerControl) => Promise<Tx[]>
+
 /**
  * @public
  */
@@ -128,6 +141,22 @@ export interface Trigger extends Doc {
   trigger: Resource<TriggerFunc>
 }
 
+/**
+ * @public
+ */
+export interface AsyncTrigger extends Doc {
+  trigger: Resource<AsyncTriggerFunc>
+  classes: Ref<Class<Doc>>[]
+}
+
+/**
+ * @public
+ */
+export interface AsyncTriggerState extends Doc {
+  tx: TxCUD<Doc>
+  message: string
+}
+
 /**
  * @public
  */