Communication indexer (#9523)

dev/docker-compose.yaml

@@ -310,6 +310,9 @@ services:
     image: hardcoreeng/fulltext
     extra_hosts:
       - 'huly.local:host-gateway'
+    depends_on:
+      elastic:
+        condition: service_healthy
     restart: unless-stopped
     links:
       - elastic

models/core/src/migration.ts

@@ -974,6 +974,13 @@ export const coreOperation: MigrateOperation = {
         state: 'clean-old-model',
         mode: 'upgrade',
         func: cleanOldModel
+      },
+      {
+        state: 'reindex-after-elastic-mapping-change',
+        mode: 'upgrade',
+        func: async (client) => {
+          await client.fullReindex()
+        }
       }
       // ,
       // {

packages/core/src/storage.ts

@@ -254,7 +254,7 @@ export interface SearchResultDoc {
   description?: string
   emojiIcon?: string
   score?: number
-  doc: Pick<Doc, '_id' | '_class'>
+  doc: Pick<Doc, '_id' | '_class' | 'createdOn'> & Partial<Pick<AttachedDoc, 'attachedTo' | 'attachedToClass'>>
 }
 
 /**

packages/model/src/migration.ts

@@ -118,6 +118,7 @@ export interface MigrationClient {
 
   wsIds: WorkspaceIds
 
+  fullReindex: () => Promise<void>
   reindex: (domain: Domain, classes: Ref<Class<Doc>>[]) => Promise<void>
   readonly logger: ModelLogger
   readonly ctx: MeasureContext

pods/fulltext/package.json

@@ -75,6 +75,8 @@
     "@hcengineering/server-storage": "^0.6.0",
     "@hcengineering/postgres": "^0.6.0",
     "@hcengineering/mongo": "^0.6.1",
-    "@hcengineering/kafka": "^0.6.0"
+    "@hcengineering/kafka": "^0.6.0",
+    "@hcengineering/communication-server": "^0.1.0",
+    "@hcengineering/communication-sdk-types": "^0.1.0"
   }
 }

pods/fulltext/src/manager.ts

@@ -6,6 +6,7 @@ import type {
   Tx,
   TxCreateDoc,
   TxCUD,
+  TxDomainEvent,
   Version,
   WorkspaceInfoWithStatus,
   WorkspaceUuid
@@ -29,8 +30,9 @@ import {
   type QueueWorkspaceReindexMessage,
   type StorageAdapter
 } from '@hcengineering/server-core'
-import { type FulltextDBConfiguration } from '@hcengineering/server-indexer'
+import { type QueueSourced, type FulltextDBConfiguration } from '@hcengineering/server-indexer'
 import { generateToken } from '@hcengineering/server-token'
+import { type Event } from '@hcengineering/communication-sdk-types'
 
 import { WorkspaceIndexer } from './workspace'
 
@@ -123,7 +125,7 @@ export class WorkspaceManager {
     )
 
     let txMessages: number = 0
-    this.txConsumer = this.opt.queue.createConsumer<TxCUD<Doc>>(
+    this.txConsumer = this.opt.queue.createConsumer<TxCUD<Doc> | TxDomainEvent<QueueSourced<Event>>>(
       this.ctx,
       QueueTopic.Tx,
       this.opt.queue.getClientId(),
@@ -136,12 +138,15 @@ export class WorkspaceManager {
 
         txMessages += msg.length
 
-        await this.processDocuments(msg, control)
+        await this.processTransactions(msg, control)
       }
     )
   }
 
-  private async processDocuments (msg: ConsumerMessage<TxCUD<Doc<Space>>>[], control: ConsumerControl): Promise<void> {
+  private async processTransactions (
+    msg: ConsumerMessage<TxCUD<Doc<Space>> | TxDomainEvent<QueueSourced<Event>>>[],
+    control: ConsumerControl
+  ): Promise<void> {
     for (const m of msg) {
       const ws = m.workspace
 
@@ -154,7 +159,7 @@ export class WorkspaceManager {
       }
 
       await this.withIndexer(this.ctx, ws, token, true, async (indexer) => {
-        await indexer.fulltext.processDocuments(this.ctx, m.value, control)
+        await indexer.fulltext.processTransactions(this.ctx, m.value, control)
       })
     }
   }

pods/fulltext/src/workspace.ts

@@ -38,6 +38,7 @@ import {
 import { FullTextIndexPipeline } from '@hcengineering/server-indexer'
 import { getConfig } from '@hcengineering/server-pipeline'
 import { generateToken } from '@hcengineering/server-token'
+import { Api as CommunicationApi } from '@hcengineering/communication-server'
 
 import { fulltextModelFilter } from './utils'
 
@@ -99,6 +100,15 @@ export class WorkspaceIndexer {
     const token = generateToken(systemAccountUuid, workspace.uuid, { service: 'fulltext' })
     const transactorEndpoint = await endpointProvider(token)
 
+    let communicationApi: CommunicationApi | undefined
+    if (process.env.COMMUNICATION_API_ENABLED === 'true') {
+      communicationApi = await CommunicationApi.create(ctx, workspace.uuid, dbURL, {
+        broadcast: () => {},
+        enqueue: () => {},
+        registerAsyncRequest: () => {}
+      })
+    }
+
     result.fulltext = new FullTextIndexPipeline(
       ftadapter,
       defaultAdapter,
@@ -137,6 +147,7 @@ export class WorkspaceIndexer {
           })
         }
       },
+      communicationApi,
       listener
     )
     return result

server/core/src/types.ts

@@ -380,9 +380,16 @@ export interface FullTextAdapter {
     update: Record<string, any>
   ) => Promise<TxResult>
   remove: (ctx: MeasureContext, workspace: WorkspaceUuid, id: Ref<Doc>[]) => Promise<void>
+  removeByQuery: (ctx: MeasureContext, workspace: WorkspaceUuid, query: DocumentQuery<Doc>) => Promise<void>
 
   clean: (ctx: MeasureContext, workspace: WorkspaceUuid) => Promise<void>
   updateMany: (ctx: MeasureContext, workspace: WorkspaceUuid, docs: IndexedDoc[]) => Promise<TxResult[]>
+  updateByQuery: (
+    ctx: MeasureContext,
+    workspace: WorkspaceUuid,
+    query: DocumentQuery<Doc>,
+    update: Record<string, any>
+  ) => Promise<TxResult[]>
   load: (ctx: MeasureContext, workspace: WorkspaceUuid, docs: Ref<Doc>[]) => Promise<IndexedDoc[]>
   searchString: (
     ctx: MeasureContext,

server/elastic/src/adapter.ts

@@ -39,7 +39,67 @@ function getIndexName (): string {
 }
 
 function getIndexVersion (): string {
-  return getMetadata(serverCore.metadata.ElasticIndexVersion) ?? 'v1'
+  return getMetadata(serverCore.metadata.ElasticIndexVersion) ?? 'v2'
+}
+
+const mappings = {
+  properties: {
+    fulltextSummary: {
+      type: 'text',
+      analyzer: 'rebuilt_english'
+    },
+    workspaceId: {
+      type: 'keyword',
+      index: true
+    },
+    id: {
+      type: 'keyword',
+      index: true
+    },
+    _class: {
+      type: 'keyword',
+      index: true
+    },
+    attachedTo: {
+      type: 'keyword',
+      index: true
+    },
+    attachedToClass: {
+      type: 'keyword',
+      index: true
+    },
+    space: {
+      type: 'keyword',
+      index: true
+    },
+    'core:class:Doc%createdBy': {
+      type: 'keyword',
+      index: true
+    },
+    'core:class:Doc%createdOn': {
+      type: 'date',
+      format: 'epoch_millis',
+      index: true
+    },
+    modifiedBy: {
+      type: 'keyword',
+      index: true
+    },
+    modifiedOn: {
+      type: 'date',
+      format: 'epoch_millis',
+      index: true
+    },
+    'core:class:Doc%modifiedBy': {
+      type: 'keyword',
+      index: true
+    },
+    'core:class:Doc%modifiedOn': {
+      type: 'date',
+      format: 'epoch_millis',
+      index: true
+    }
+  }
 }
 
 class ElasticAdapter implements FullTextAdapter {
@@ -100,34 +160,25 @@ class ElasticAdapter implements FullTextAdapter {
                     }
                   }
                 }
-              }
+              },
+              mappings
             }
           })
         )
+      } else {
+        await ctx.with('put-mapping', {}, () =>
+          this.client.indices.putMapping({
+            index: indexName,
+            body: mappings
+          })
+        )
       }
-
-      await ctx.with('put-mapping', {}, () =>
-        this.client.indices.putMapping({
-          index: indexName,
-          body: {
-            properties: {
-              fulltextSummary: {
-                type: 'text',
-                analyzer: 'rebuilt_english'
-              },
-              workspaceId: {
-                type: 'keyword',
-                index: true
-              }
-            }
-          }
-        })
-      )
     } catch (err: any) {
-      if (err.name !== 'ConnectionError') {
-        Analytics.handleError(err)
-        ctx.error(err)
+      if (err.name === 'ConnectionError') {
+        ctx.warn('Elastic DB is not available')
       }
+      Analytics.handleError(err)
+      ctx.error(err)
       return false
     }
     return true
@@ -164,8 +215,8 @@ class ElasticAdapter implements FullTextAdapter {
                     }
                   },
                   {
-                    match: {
-                      workspaceId: { query: workspaceId, operator: 'and' }
+                    term: {
+                      workspaceId
                     }
                   }
                 ]
@@ -185,12 +236,12 @@ class ElasticAdapter implements FullTextAdapter {
 
       if (query.spaces !== undefined) {
         filter.push({
-          terms: { 'space.keyword': query.spaces }
+          terms: this.getTerms(query.spaces, 'space')
         })
       }
       if (query.classes !== undefined) {
         filter.push({
-          terms: { '_class.keyword': query.classes }
+          terms: this.getTerms(query.classes, '_class')
         })
       }
 
@@ -200,9 +251,12 @@ class ElasticAdapter implements FullTextAdapter {
 
       if (options.scoring !== undefined) {
         const scoringTerms: any[] = options.scoring.map((scoringOption): any => {
+          const field = Object.hasOwn(mappings.properties, scoringOption.attr)
+            ? scoringOption.attr
+            : `${scoringOption.attr}.keyword`
           return {
             term: {
-              [`${scoringOption.attr}.keyword`]: {
+              [field]: {
                 value: scoringOption.value,
                 boost: scoringOption.boost
               }
@@ -258,8 +312,8 @@ class ElasticAdapter implements FullTextAdapter {
             }
           },
           {
-            match: {
-              workspaceId: { query: workspaceId, operator: 'and' }
+            term: {
+              workspaceId
             }
           }
         ],
@@ -279,11 +333,12 @@ class ElasticAdapter implements FullTextAdapter {
 
     for (const [q, v] of Object.entries(query)) {
       if (!q.startsWith('$')) {
+        const field = Object.hasOwn(mappings.properties, q) ? q : `${q}.keyword`
         if (typeof v === 'object') {
           if (v.$in !== undefined) {
             request.bool.should.push({
               terms: {
-                [q]: v.$in,
+                [field]: v.$in,
                 boost: 100.0
               }
             })
@@ -291,7 +346,7 @@ class ElasticAdapter implements FullTextAdapter {
         } else {
           request.bool.should.push({
             term: {
-              [q]: {
+              [field]: {
                 value: v,
                 boost: 100.0,
                 case_insensitive: true
@@ -335,9 +390,9 @@ class ElasticAdapter implements FullTextAdapter {
     }
   }
 
-  private getTerms (_classes: Ref<Class<Doc>>[], field: string, extra: any = {}): any {
+  private getTerms (values: string[], field: string, extra: any = {}): any {
     return {
-      [field]: _classes.map((c) => c.toLowerCase()),
+      [Object.hasOwn(mappings.properties, field) ? field : `${field}.keyword`]: values,
       ...extra
     }
   }
@@ -415,6 +470,64 @@ class ElasticAdapter implements FullTextAdapter {
     return []
   }
 
+  async updateByQuery (
+    ctx: MeasureContext,
+    workspaceId: WorkspaceUuid,
+    query: DocumentQuery<Doc>,
+    update: Record<string, any>
+  ): Promise<TxResult[]> {
+    const elasticQuery: any = {
+      bool: {
+        must: [
+          {
+            term: {
+              workspaceId
+            }
+          }
+        ]
+      }
+    }
+
+    for (const [q, v] of Object.entries(query)) {
+      if (!q.startsWith('$')) {
+        if (typeof v === 'object') {
+          if (v.$in !== undefined) {
+            elasticQuery.bool.must.push({
+              terms: {
+                [Object.hasOwn(mappings.properties, q) ? q : `${q}.keyword`]: v.$in
+              }
+            })
+          }
+        } else {
+          elasticQuery.bool.must.push({
+            term: {
+              [Object.hasOwn(mappings.properties, q) ? q : `${q}.keyword`]: {
+                value: v
+              }
+            }
+          })
+        }
+      }
+    }
+
+    await this.client.updateByQuery({
+      type: '_doc',
+      index: this.indexName,
+      body: {
+        query: elasticQuery,
+        script: {
+          source:
+            'for(int i = 0; i < params.updateFields.size(); i++) { ctx._source[params.updateFields[i].key] = params.updateFields[i].value }',
+          params: {
+            updateFields: Object.entries(update).map(([key, value]) => ({ key, value }))
+          },
+          lang: 'painless'
+        }
+      }
+    })
+    return []
+  }
+
   async remove (ctx: MeasureContext, workspaceId: WorkspaceUuid, docs: Ref<Doc>[]): Promise<void> {
     try {
       while (docs.length > 0) {
@@ -434,8 +547,8 @@ class ElasticAdapter implements FullTextAdapter {
                       }
                     },
                     {
-                      match: {
-                        workspaceId: { query: workspaceId, operator: 'and' }
+                      term: {
+                        workspaceId
                       }
                     }
                   ]
@@ -455,6 +568,56 @@ class ElasticAdapter implements FullTextAdapter {
     }
   }
 
+  async removeByQuery (ctx: MeasureContext, workspaceId: WorkspaceUuid, query: DocumentQuery<Doc>): Promise<void> {
+    const elasticQuery: any = {
+      bool: {
+        must: [
+          {
+            term: {
+              workspaceId
+            }
+          }
+        ]
+      }
+    }
+
+    for (const [q, v] of Object.entries(query)) {
+      if (!q.startsWith('$')) {
+        if (typeof v === 'object') {
+          if (v.$in !== undefined) {
+            elasticQuery.bool.must.push({
+              terms: {
+                [Object.hasOwn(mappings.properties, q) ? q : `${q}.keyword`]: v.$in
+              }
+            })
+          }
+        } else {
+          elasticQuery.bool.must.push({
+            term: {
+              [Object.hasOwn(mappings.properties, q) ? q : `${q}.keyword`]: {
+                value: v
+              }
+            }
+          })
+        }
+      }
+    }
+    try {
+      await this.client.deleteByQuery({
+        type: '_doc',
+        index: this.indexName,
+        body: {
+          query: elasticQuery
+        }
+      })
+    } catch (e: any) {
+      if (e instanceof esErr.ResponseError && e.meta.statusCode === 404) {
+        return
+      }
+      throw e
+    }
+  }
+
   async clean (ctx: MeasureContext, workspaceId: WorkspaceUuid): Promise<void> {
     try {
       await this.client.deleteByQuery(
@@ -466,8 +629,8 @@ class ElasticAdapter implements FullTextAdapter {
               bool: {
                 must: [
                   {
-                    match: {
-                      workspaceId: { query: workspaceId, operator: 'and' }
+                    term: {
+                      workspaceId
                     }
                   }
                 ]
@@ -500,8 +663,8 @@ class ElasticAdapter implements FullTextAdapter {
                 }
               },
               {
-                match: {
-                  workspaceId: { query: workspaceId, operator: 'and' }
+                term: {
+                  workspaceId
                 }
               }
             ]

server/indexer/package.json

@@ -40,12 +40,19 @@
     "@hcengineering/server-core": "^0.6.1",
     "@hcengineering/server-token": "^0.6.11",
     "@hcengineering/text": "^0.6.5",
+    "@hcengineering/text-markdown": "^0.6.0",
     "@hcengineering/analytics": "^0.6.0",
     "@hcengineering/query": "^0.6.12",
     "@hcengineering/contact": "^0.6.24",
     "@hcengineering/attachment": "^0.6.14",
+    "@hcengineering/card": "^0.6.0",
     "@hcengineering/drive": "^0.6.0",
     "fast-equals": "^5.2.2",
-    "@hcengineering/storage": "^0.6.0"
+    "@hcengineering/storage": "^0.6.0",
+    "@hcengineering/communication-rest-client": "^0.1.0",
+    "@hcengineering/communication-sdk-types": "^0.1.0",
+    "@hcengineering/communication-shared": "^0.1.0",
+    "@hcengineering/communication-types": "^0.1.0",
+    "@hcengineering/communication-yaml": "^0.1.0"
   }
 }

server/indexer/src/indexer/indexer.ts

@@ -30,8 +30,10 @@ import core, {
   type MeasureContext,
   type ModelDb,
   type Ref,
+  SortingOrder,
   type Space,
   type TxCUD,
+  type TxDomainEvent,
   TxProcessor,
   type WorkspaceIds,
   type WorkspaceUuid,
@@ -58,15 +60,53 @@ import type {
 } from '@hcengineering/server-core'
 import { RateLimiter, SessionDataImpl } from '@hcengineering/server-core'
 import { jsonToText, markupToJSON, markupToText } from '@hcengineering/text'
+import card, { type Card } from '@hcengineering/card'
 import { findSearchPresenter, updateDocWithPresenter } from '../mapper'
 import { type FullTextPipeline } from './types'
-import { createIndexedDoc, getContent } from './utils'
+import { blobPseudoClass, createIndexedDoc, createIndexedDocFromMessage, getContent, messagePseudoClass } from './utils'
+import {
+  type ServerApi as CommunicationApi,
+  type SessionData as CommunicationSession,
+  type CreateMessageEvent,
+  type UpdatePatchEvent,
+  type RemovePatchEvent,
+  MessageEventType,
+  type Event,
+  CardEventType,
+  type UpdateCardTypeEvent,
+  type EventType,
+  type BlobPatchEvent,
+  type LinkPreviewPatchEvent,
+  type RemoveCardEvent
+} from '@hcengineering/communication-sdk-types'
+import { type AttachedBlob, type CardID, type Message, type MessageID } from '@hcengineering/communication-types'
+import { parseYaml } from '@hcengineering/communication-yaml'
+import { applyPatches } from '@hcengineering/communication-shared'
+import { markdownToMarkup } from '@hcengineering/text-markdown'
 
 export * from './types'
 export * from './utils'
 
+const printThresholdMs = 2500
+
 const textLimit = 500 * 1024
 
+const messageGroupsLimit = 100
+const messagesLimit = 1000
+
+// Inner presentation in message queue differs from sdk-types,
+// also date is always filled at the output queue
+export type QueueSourced<T extends Event> = Omit<T, 'date'> & { date: string }
+
+type IndexableCommunicationEvent =
+  | QueueSourced<CreateMessageEvent>
+  | QueueSourced<UpdatePatchEvent>
+  | QueueSourced<BlobPatchEvent>
+  | QueueSourced<LinkPreviewPatchEvent>
+  | QueueSourced<RemovePatchEvent>
+  | QueueSourced<UpdateCardTypeEvent>
+  | QueueSourced<RemoveCardEvent>
+
 // Global Memory management configuration
 
 /**
@@ -167,6 +207,8 @@ export class FullTextIndexPipeline implements FullTextPipeline {
 
   contexts: Map<Ref<Class<Doc>>, FullTextSearchContext>
 
+  communicationSession: CommunicationSession
+
   constructor (
     readonly fulltextAdapter: FullTextAdapter,
     private readonly storage: DbAdapter,
@@ -177,9 +219,11 @@ export class FullTextIndexPipeline implements FullTextPipeline {
     readonly storageAdapter: StorageAdapter,
     readonly contentAdapter: ContentTextAdapter,
     readonly broadcastUpdate: (ctx: MeasureContext, classes: Ref<Class<Doc>>[]) => void,
+    readonly communicationApi?: CommunicationApi,
     readonly listener?: FulltextListener
   ) {
     this.contexts = new Map(model.findAllSync(core.class.FullTextSearchContext, {}).map((it) => [it.toClass, it]))
+    this.communicationSession = { account: systemAccount, asyncData: [] }
   }
 
   async getIndexClassess (): Promise<{ domain: Domain, classes: Ref<Class<Doc>>[] }[]> {
@@ -217,6 +261,8 @@ export class FullTextIndexPipeline implements FullTextPipeline {
     ctx.warn('verify document structure', { workspace: this.workspace.uuid })
 
     let processed = 0
+    let processedCommunication = 0
+    let hasCards = false
     await ctx.with('reindex-domain', { domain }, async (ctx) => {
       // Iterate over all domain documents and add appropriate entries
       const allDocs = this.storage.rawFind(ctx, domain)
@@ -238,6 +284,9 @@ export class FullTextIndexPipeline implements FullTextPipeline {
               // Skip non indexable classes
               continue
             }
+            if (!hasCards && this.hierarchy.isDerived(v, card.class.Card)) {
+              hasCards = true
+            }
 
             await this.indexDocuments(ctx, v, values, pushQueue)
             await control?.heartbeat()
@@ -250,7 +299,7 @@ export class FullTextIndexPipeline implements FullTextPipeline {
           // Find the next threshold to print
 
           const now = platformNow()
-          if (now - lastPrint > 2500) {
+          if (now - lastPrint > printThresholdMs) {
             ctx.info('processed', { processed, elapsed: Math.round(now - lastPrint), domain })
             lastPrint = now
           }
@@ -261,8 +310,19 @@ export class FullTextIndexPipeline implements FullTextPipeline {
       } finally {
         await allDocs.close()
       }
+      if (hasCards) {
+        await ctx.with('reindex-communication', {}, async (ctx) => {
+          try {
+            const pushQueue = new ElasticPushQueue(this.fulltextAdapter, this.workspace, ctx, control)
+            processedCommunication = await this.indexCommunication(ctx, control, pushQueue)
+            await pushQueue.waitProcessing()
+          } catch (err: any) {
+            ctx.error('failed to restore index state', { err })
+          }
+        })
+      }
     })
-    ctx.warn('reinex done', { domain, processed })
+    ctx.info('reindex done', { domain, processed, processedCommunication })
   }
 
   async dropWorkspace (control?: ConsumerControl): Promise<void> {
@@ -466,14 +526,193 @@ export class FullTextIndexPipeline implements FullTextPipeline {
     await rateLimit.waitProcessing()
   }
 
-  public async processDocuments (ctx: MeasureContext, result: TxCUD<Doc>[], control: ConsumerControl): Promise<void> {
+  async indexCommunication (
+    ctx: MeasureContext,
+    control: ConsumerControl | undefined,
+    pushQueue: ElasticPushQueue
+  ): Promise<number> {
+    const communicationApi = this.communicationApi
+    if (communicationApi === undefined) {
+      return 0
+    }
+    let processed = 0
+    const cardsInfo = new Map<CardID, { space: Ref<Space>, _class: Ref<Class<Doc>> }>()
+    const rateLimit = new RateLimiter(10)
+    let lastPrint = 0
+    await ctx.with('process-message-groups', {}, async (ctx) => {
+      let groups = await communicationApi.findMessagesGroups(this.communicationSession, {
+        limit: messageGroupsLimit,
+        order: SortingOrder.Ascending
+      })
+      while (groups.length > 0) {
+        if (this.cancelling) {
+          return processed
+        }
+        for (const group of groups) {
+          if (control !== undefined) {
+            await control.heartbeat()
+          }
+          try {
+            let cardInfo = cardsInfo.get(group.cardId)
+            if (cardInfo === undefined) {
+              const cardDoc = await this.storage.findAll(ctx, card.class.Card, { _id: group.cardId }, { limit: 1 })
+              if (cardDoc.length !== 1) {
+                continue
+              }
+              cardInfo = { space: cardDoc[0].space, _class: cardDoc[0]._class }
+              cardsInfo.set(group.cardId, cardInfo)
+            }
+            const blob = await this.storageAdapter.read(ctx, this.workspace, group.blobId)
+            const messagesFile = Buffer.concat(blob as any).toString()
+            const messagesParsedFile = parseYaml(messagesFile)
+            let patchedMessages
+            if (group.patches !== undefined && group.patches.length > 0) {
+              const patchesByMessage = groupByArray(group.patches, (it) => it.messageId)
+              patchedMessages = messagesParsedFile.messages.map((message) => {
+                const patches = patchesByMessage.get(message.id) ?? []
+                if (patches.length === 0) {
+                  return message
+                } else {
+                  return applyPatches(message, patches)
+                }
+              })
+            } else {
+              patchedMessages = messagesParsedFile.messages
+            }
+            for (const message of patchedMessages) {
+              if (message.removed) {
+                continue
+              }
+              await rateLimit.exec(async () => {
+                await this.processCommunicationMessage(
+                  ctx,
+                  pushQueue,
+                  group.cardId,
+                  cardInfo.space,
+                  cardInfo._class,
+                  message
+                )
+              })
+              processed += 1
+              const now = platformNow()
+              if (now - lastPrint > printThresholdMs) {
+                ctx.info('processed', { processedCommunication: processed, elapsed: Math.round(now - lastPrint) })
+                lastPrint = now
+              }
+            }
+          } catch (err: any) {
+            ctx.error('Failed to process message group', {
+              cardId: group.cardId,
+              blobId: group.blobId,
+              error: err
+            })
+            Analytics.handleError(err)
+          }
+        }
+        if (this.cancelling) {
+          return processed
+        }
+        groups = await communicationApi.findMessagesGroups(this.communicationSession, {
+          limit: messageGroupsLimit,
+          order: SortingOrder.Ascending,
+          fromDate: {
+            greater: groups[groups.length - 1].toDate
+          }
+        })
+      }
+    })
+    await ctx.with('process-messages', {}, async (ctx) => {
+      let messages = await communicationApi.findMessages(this.communicationSession, {
+        links: true,
+        files: true,
+        limit: messagesLimit,
+        order: SortingOrder.Ascending
+      })
+      while (messages.length > 0) {
+        for (const message of messages) {
+          if (control !== undefined) {
+            await control.heartbeat()
+          }
+          try {
+            let cardInfo = cardsInfo.get(message.cardId)
+            if (cardInfo === undefined) {
+              const cardDoc = await this.storage.findAll(ctx, card.class.Card, { _id: message.cardId }, { limit: 1 })
+              if (cardDoc.length !== 1) {
+                continue
+              }
+              cardInfo = { space: cardDoc[0].space, _class: cardDoc[0]._class }
+              cardsInfo.set(message.cardId, cardInfo)
+            }
+            if (this.cancelling) {
+              return processed
+            }
+            await rateLimit.exec(async () => {
+              await this.processCommunicationMessage(
+                ctx,
+                pushQueue,
+                message.cardId,
+                cardInfo.space,
+                cardInfo._class,
+                message
+              )
+            })
+          } catch (err: any) {
+            ctx.error('Failed to processed message', {
+              cardId: message.cardId,
+              id: message.id,
+              error: err
+            })
+          }
+          processed += 1
+          const now = platformNow()
+          if (now - lastPrint > printThresholdMs) {
+            ctx.info('processed', { processedCommunication: processed, elapsed: Math.round(now - lastPrint) })
+            lastPrint = now
+          }
+        }
+        messages = await communicationApi.findMessages(this.communicationSession, {
+          links: true,
+          files: true,
+          limit: messagesLimit,
+          order: SortingOrder.Ascending,
+          created: {
+            greater: messages[messages.length - 1].created
+          }
+        })
+      }
+    })
+    await rateLimit.waitProcessing()
+    return processed
+  }
+
+  public async processTransactions (
+    ctx: MeasureContext,
+    result: (TxCUD<Doc> | TxDomainEvent<QueueSourced<Event>>)[],
+    control: ConsumerControl
+  ): Promise<void> {
     const contextData = this.createContextData()
     ctx.contextData = contextData
-    // Find documents matching query
+
+    const indexableCommunicationEventTypes: Array<EventType> = [
+      MessageEventType.CreateMessage,
+      MessageEventType.UpdatePatch,
+      MessageEventType.BlobPatch,
+      MessageEventType.LinkPreviewPatch,
+      MessageEventType.RemovePatch,
+      CardEventType.UpdateCardType,
+      CardEventType.RemoveCard
+    ]
+
+    const docEvents = result.filter((tx) => tx._class !== core.class.TxDomainEvent) as TxCUD<Doc>[]
+    const messageEvents = result.filter(
+      (tx) =>
+        tx._class === core.class.TxDomainEvent &&
+        (tx as TxDomainEvent<any>).domain === 'communication' &&
+        indexableCommunicationEventTypes.includes((tx as TxDomainEvent<QueueSourced<Event>>).event.type)
+    ) as any as TxDomainEvent<IndexableCommunicationEvent>[]
 
     // We need to update hierarchy and local model if required.
-
-    for (const tx of result) {
+    for (const tx of docEvents) {
       try {
         this.hierarchy.tx(tx)
         const domain = this.hierarchy.findDomain(tx.objectClass)
@@ -486,7 +725,7 @@ export class FullTextIndexPipeline implements FullTextPipeline {
       }
     }
 
-    const byClass = groupByArray<TxCUD<Doc>, Ref<Class<Doc>>>(result, (it) => it.objectClass)
+    const byClass = groupByArray<TxCUD<Doc>, Ref<Class<Doc>>>(docEvents, (it) => it.objectClass)
 
     const pushQueue = new ElasticPushQueue(this.fulltextAdapter, this.workspace, ctx, control)
 
@@ -509,6 +748,16 @@ export class FullTextIndexPipeline implements FullTextPipeline {
       }
     }
 
+    const messagesByCardId = groupByArray(messageEvents, (e) => e.event.cardId)
+    for (const [cardId, txes] of messagesByCardId) {
+      try {
+        await this.processCommunicationEvents(ctx, pushQueue, cardId, txes, toRemove)
+      } catch (err: any) {
+        ctx.error('failed to index communication', { err, cardId })
+        Analytics.handleError(err)
+      }
+    }
+
     try {
       if (toRemove.length !== 0) {
         // We need to add broadcast information
@@ -529,6 +778,140 @@ export class FullTextIndexPipeline implements FullTextPipeline {
     this.scheduleBroadcast()
   }
 
+  private async processCommunicationEvents (
+    ctx: MeasureContext,
+    pushQueue: ElasticPushQueue,
+    cardId: CardID,
+    txes: TxDomainEvent<IndexableCommunicationEvent>[],
+    toRemove: { _id: Ref<Doc>, _class: Ref<Class<Doc>> }[]
+  ): Promise<void> {
+    const communicationApi = this.communicationApi
+    if (communicationApi === undefined) {
+      return
+    }
+    const getMessage = async (cardId: CardID, msgId: MessageID): Promise<Message | undefined> => {
+      const messages = await communicationApi.findMessages(this.communicationSession, {
+        card: cardId,
+        id: msgId,
+        links: true,
+        files: true
+      })
+      if (messages.length === 1) {
+        return messages[0]
+      }
+      const messagesGroups = await communicationApi.findMessagesGroups(this.communicationSession, {
+        card: cardId,
+        messageId: msgId
+      })
+      if (messagesGroups.length !== 1) {
+        return undefined
+      }
+      const group = messagesGroups[0]
+      const blob = await this.storageAdapter.read(ctx, this.workspace, group.blobId)
+      const messagesFile = Buffer.concat(blob as any).toString()
+      const messagesParsedFile = parseYaml(messagesFile)
+      const message = messagesParsedFile.messages.find((m) => m.id === msgId)
+      if (group.patches === undefined || message === undefined) {
+        return message
+      }
+      const relevantPatches = group.patches.filter((p) => p.messageId === msgId)
+      if (relevantPatches.length === 0) {
+        return message
+      } else {
+        return applyPatches(message, relevantPatches)
+      }
+    }
+    const cardDoc = (await this.storage.findAll(ctx, card.class.Card, { _id: cardId }))[0]
+    // If message was already fully replaced, other transactions can skip the message
+    const messagesUpdated = new Set<MessageID>()
+    for (const tx of txes) {
+      if (
+        [MessageEventType.CreateMessage, MessageEventType.UpdatePatch, MessageEventType.LinkPreviewPatch].includes(
+          tx.event.type as any
+        )
+      ) {
+        const event = tx.event as
+          | QueueSourced<CreateMessageEvent>
+          | QueueSourced<UpdatePatchEvent>
+          | QueueSourced<LinkPreviewPatchEvent>
+        if (event.messageId === undefined) {
+          continue
+        }
+        if (messagesUpdated.has(event.messageId)) {
+          continue
+        }
+        const message = await getMessage(cardId, event.messageId)
+        if (message === undefined) {
+          continue
+        }
+        await this.processCommunicationMessage(ctx, pushQueue, cardDoc._id, cardDoc.space, cardDoc._class, message)
+        messagesUpdated.add(event.messageId)
+      } else if (tx.event.type === MessageEventType.BlobPatch) {
+        const event = tx.event
+        if (messagesUpdated.has(event.messageId)) {
+          continue
+        }
+        for (const operation of event.operations) {
+          if (operation.opcode === 'attach' || operation.opcode === 'set' || operation.opcode === 'update') {
+            for (const blobData of operation.blobs) {
+              const attachedBlob = Object.assign(blobData, {
+                creator: event.socialId,
+                created: new Date(Date.parse(event.date))
+              })
+              await this.processCommunicationBlob(
+                ctx,
+                pushQueue,
+                {
+                  id: `${event.messageId}@${cardDoc._id}` as any,
+                  _class: [messagePseudoClass],
+                  space: cardDoc.space,
+                  attachedTo: cardDoc._id
+                },
+                attachedBlob as AttachedBlob
+              )
+            }
+          } else if (operation.opcode === 'detach') {
+            for (const blobId of operation.blobIds) {
+              toRemove.push({
+                _id: `${blobId}@${cardDoc._id}` as Ref<Doc>,
+                _class: blobPseudoClass
+              })
+            }
+          }
+        }
+      } else if (tx.event.type === MessageEventType.RemovePatch) {
+        const event = tx.event
+        messagesUpdated.add(event.messageId)
+        await this.fulltextAdapter.removeByQuery(ctx, this.workspace.uuid, {
+          _class: blobPseudoClass,
+          attachedTo: `${event.messageId}@${event.cardId}` as Ref<Doc>
+        })
+        toRemove.push({
+          _id: `${event.messageId}@${event.cardId}` as any,
+          _class: messagePseudoClass
+        })
+      } else if (tx.event.type === CardEventType.UpdateCardType) {
+        const event = tx.event
+        await this.fulltextAdapter.updateByQuery(
+          ctx,
+          this.workspace.uuid,
+          { _class: messagePseudoClass, attachedTo: event.cardId },
+          { attachedToClass: event.cardType }
+        )
+      } else if (tx.event.type === CardEventType.RemoveCard) {
+        const event = tx.event
+        await this.fulltextAdapter.removeByQuery(ctx, this.workspace.uuid, {
+          _class: messagePseudoClass,
+          attachedTo: event.cardId
+        })
+        await this.fulltextAdapter.removeByQuery(ctx, this.workspace.uuid, {
+          _class: blobPseudoClass,
+          attachedToCard: event.cardId
+        })
+      }
+    }
+  }
+
   private async loadDocsFromTx (
     values: TxCUD<Doc<Space>>[],
     toRemove: { _id: Ref<Doc<Space>>, _class: Ref<Class<Doc>> }[],
@@ -637,20 +1020,7 @@ export class FullTextIndexPipeline implements FullTextPipeline {
           return
         }
       }
-      const docInfo: Blob | undefined = await this.storageAdapter.stat(ctx, this.workspace, ref)
-      if (docInfo !== undefined && docInfo.size < 30 * 1024 * 1024) {
-        // We have blob, we need to decode it to string.
-        const contentType = (docInfo.contentType ?? '').split(';')[0]
-
-        if (
-          (contentType.includes('text/') && contentType !== 'text/rtf') ||
-          contentType.includes('application/vnd.github.VERSION.diff')
-        ) {
-          await this.handleTextBlob(ctx, docInfo, indexedDoc)
-        } else if (isBlobAllowed(contentType)) {
-          await this.handleBlob(ctx, docInfo, indexedDoc)
-        }
-      }
+      await this.handleBlobRef(ctx, ref, indexedDoc)
     } catch (err: any) {
       ctx.warn('faild to process text content', {
         id: doc._id,
@@ -662,6 +1032,104 @@ export class FullTextIndexPipeline implements FullTextPipeline {
     }
   }
 
+  @withContext('process-communication-message')
+  private async processCommunicationMessage (
+    ctx: MeasureContext<any>,
+    pushQueue: ElasticPushQueue,
+    cardId: CardID,
+    cardSpace: Ref<Space>,
+    cardClass: Ref<Class<Card>>,
+    message: Pick<
+    Message,
+    'id' | 'edited' | 'created' | 'creator' | 'content' | 'extra' | 'blobs' | 'thread' | 'linkPreviews'
+    >
+  ): Promise<void> {
+    const indexedDoc = createIndexedDocFromMessage(cardId, cardSpace, cardClass, message)
+    const markup = markdownToMarkup(message.content)
+    let textContent = jsonToText(markup)
+    textContent = textContent
+      .split(/ +|\t+|\f+/)
+      .filter((it) => it)
+      .join(' ')
+      .split(/\n\n+/)
+      .join('\n')
+    indexedDoc.fulltextSummary = textContent
+    for (const linkPreview of message.linkPreviews) {
+      if (linkPreview.title !== undefined) {
+        indexedDoc.fulltextSummary += '\n' + linkPreview.title
+      }
+      if (linkPreview.siteName !== undefined) {
+        indexedDoc.fulltextSummary += '\n' + linkPreview.siteName
+      }
+      if (linkPreview.description !== undefined) {
+        indexedDoc.fulltextSummary += '\n' + linkPreview.description
+      }
+    }
+    if (this.listener?.onIndexing !== undefined) {
+      await this.listener.onIndexing(indexedDoc)
+    }
+    await pushQueue.push(indexedDoc)
+    for (const blob of message.blobs) {
+      await this.processCommunicationBlob(ctx, pushQueue, indexedDoc, blob)
+    }
+  }
+
+  @withContext('process-communication-blob')
+  private async processCommunicationBlob (
+    ctx: MeasureContext<any>,
+    pushQueue: ElasticPushQueue,
+    parentDoc: { id: Ref<Doc>, _class: Ref<Class<Doc>>[], space: Ref<Space>, attachedTo?: Ref<Doc> },
+    blob: AttachedBlob
+  ): Promise<void> {
+    try {
+      const indexedDoc: IndexedDoc = {
+        id: `${blob.blobId}@${parentDoc.attachedTo}` as any,
+        _class: [`${card.class.Card}%blob` as Ref<Class<Doc>>],
+        space: parentDoc.space,
+        [docKey('createdOn', core.class.Doc)]: blob.created.getTime(),
+        [docKey('createdBy', core.class.Doc)]: blob.creator,
+        modifiedBy: blob.creator,
+        modifiedOn: blob.created.getTime(),
+        attachedTo: parentDoc.id,
+        attachedToClass: parentDoc._class[0],
+        searchTitle: blob.fileName,
+        searchShortTitle: blob.fileName,
+        attachedToCard: parentDoc.attachedTo
+      }
+      indexedDoc.fulltextSummary = ''
+      await this.handleBlobRef(ctx, blob.blobId, indexedDoc, blob.mimeType)
+      if (this.listener?.onIndexing !== undefined) {
+        await this.listener.onIndexing(indexedDoc)
+      }
+      await pushQueue.push(indexedDoc)
+    } catch (err: any) {
+      Analytics.handleError(err)
+      ctx.error('failed to handle blob', { err, _id: blob.blobId, workspace: this.workspace.uuid })
+    }
+  }
+
+  private async handleBlobRef (
+    ctx: MeasureContext<any>,
+    ref: Ref<Blob>,
+    indexedDoc: IndexedDoc,
+    defaultContentType: string = ''
+  ): Promise<void> {
+    const docInfo: Blob | undefined = await this.storageAdapter.stat(ctx, this.workspace, ref)
+    if (docInfo !== undefined && docInfo.size < 30 * 1024 * 1024) {
+      // We have blob, we need to decode it to string.
+      const contentType = (docInfo.contentType ?? defaultContentType).split(';')[0]
+
+      if (
+        (contentType.includes('text/') && contentType !== 'text/rtf') ||
+        contentType.includes('application/vnd.github.VERSION.diff')
+      ) {
+        await this.handleTextBlob(ctx, docInfo, indexedDoc)
+      } else if (isBlobAllowed(contentType)) {
+        await this.handleBlob(ctx, docInfo, indexedDoc)
+      }
+    }
+  }
+
   private async handleBlob (ctx: MeasureContext<any>, docInfo: Blob | undefined, indexedDoc: IndexedDoc): Promise<void> {
     if (docInfo !== undefined) {
       const contentType = (docInfo.contentType ?? '').split(';')[0]

server/indexer/src/indexer/utils.ts

@@ -13,10 +13,11 @@
 // limitations under the License.
 //
 
-import {
+import core, {
   type AnyAttribute,
   type Class,
   type Doc,
+  docKey,
   type FullTextSearchContext,
   getFullTextContext,
   type Hierarchy,
@@ -25,6 +26,8 @@ import {
 } from '@hcengineering/core'
 import { type IndexedDoc } from '@hcengineering/server-core'
 import { type FullTextPipeline } from './types'
+import { type Message } from '@hcengineering/communication-types'
+import cardPlugin, { type Card } from '@hcengineering/card'
 
 export { docKey, isFullTextAttribute } from '@hcengineering/core'
 
@@ -103,3 +106,31 @@ export function createIndexedDoc (doc: Doc, mixins: Ref<Class<Doc>>[] | undefine
   }
   return indexedDoc
 }
+
+export const messagePseudoClass = `${cardPlugin.class.Card}%message` as Ref<Class<Doc>>
+export const blobPseudoClass = `${cardPlugin.class.Card}%blob` as Ref<Class<Doc>>
+
+/**
+ * @public
+ */
+export function createIndexedDocFromMessage (
+  cardId: Ref<Card>,
+  cardSpace: Ref<Space>,
+  cardClass: Ref<Class<Card>>,
+  message: Pick<Message, 'id' | 'edited' | 'created' | 'creator'>
+): IndexedDoc {
+  const modifiedDate = message.edited ?? message.created
+  const modifiedOn = modifiedDate.getTime()
+  const indexedDoc = {
+    id: `${message.id}@${cardId}` as any,
+    _class: [messagePseudoClass],
+    space: cardSpace,
+    [docKey('createdOn', core.class.Doc)]: message.created.getTime(),
+    [docKey('createdBy', core.class.Doc)]: message.creator,
+    modifiedBy: message.creator,
+    modifiedOn,
+    attachedTo: cardId,
+    attachedToClass: cardClass
+  }
+  return indexedDoc
+}

server/indexer/src/mapper.ts

@@ -139,7 +139,10 @@ export function mapSearchResultDoc (hierarchy: Hierarchy, raw: IndexedDoc): Sear
     shortTitle: raw.searchShortTitle,
     doc: {
       _id: raw.id,
-      _class: raw._class[0]
+      _class: raw._class[0],
+      createdOn: raw.createdOn,
+      attachedTo: raw.attachedTo,
+      attachedToClass: raw.attachedToClass
     },
     score: raw._score
   }

server/middleware/package.json

@@ -41,6 +41,7 @@
     "@hcengineering/server-preference": "^0.6.0",
     "@hcengineering/query": "^0.6.12",
     "@hcengineering/analytics": "^0.6.0",
+    "@hcengineering/card": "^0.6.0",
     "fast-equals": "^5.2.2"
   }
 }

server/middleware/src/fulltext.ts

@@ -36,6 +36,7 @@ import core, {
 } from '@hcengineering/core'
 import type { IndexedDoc, Middleware, MiddlewareCreator, PipelineContext } from '@hcengineering/server-core'
 import { BaseMiddleware } from '@hcengineering/server-core'
+import card from '@hcengineering/card'
 /**
  * @public
  */
@@ -163,6 +164,10 @@ export class FullTextMiddleware extends BaseMiddleware implements Middleware {
           }
         }
       }
+      if (this.context.hierarchy.isDerived(baseClass, card.class.Card)) {
+        // Using Card as base class because messages are the same for any card subclass
+        childClasses.add(`${card.class.Card}%message` as Ref<Class<Doc>>)
+      }
     } catch (err: any) {
       Analytics.handleError(err)
     }

server/tool/src/upgrade.ts

@@ -127,6 +127,10 @@ export class MigrateClientImpl implements MigrationClient {
     await this.lowLevel.rawDeleteMany(domain, query)
   }
 
+  async fullReindex (): Promise<void> {
+    await this.queue.send(this.wsIds.uuid, [workspaceEvents.fullReindex()])
+  }
+
   async reindex (domain: Domain, classes: Ref<Class<Doc>>[]): Promise<void> {
     await this.queue.send(this.wsIds.uuid, [workspaceEvents.reindex(domain, classes)])
   }

tests/docker-compose.yaml

@@ -104,7 +104,7 @@ services:
       - discovery.type=single-node
       - ES_JAVA_OPTS=-Xms1024m -Xmx1024m
     healthcheck:
-      interval: 20s
+      interval: 5s
       retries: 10
       test: curl -s http://localhost:9200/_cluster/health | grep -vq '"status":"red"'
   account:
@@ -257,6 +257,8 @@ services:
         condition: service_started
       cockroach:
         condition: service_started
+      elastic:
+        condition: service_healthy
     links:
       - elastic
       - mongodb