diff --git a/dev/docker-compose.yaml b/dev/docker-compose.yaml index 0e94610b74..095780d5f6 100644 --- a/dev/docker-compose.yaml +++ b/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 diff --git a/models/core/src/migration.ts b/models/core/src/migration.ts index a6ff27c61a..e259d04abc 100644 --- a/models/core/src/migration.ts +++ b/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() + } } // , // { diff --git a/packages/core/src/storage.ts b/packages/core/src/storage.ts index ae7ec7643c..4f0b69348c 100644 --- a/packages/core/src/storage.ts +++ b/packages/core/src/storage.ts @@ -254,7 +254,7 @@ export interface SearchResultDoc { description?: string emojiIcon?: string score?: number - doc: Pick + doc: Pick & Partial> } /** diff --git a/packages/model/src/migration.ts b/packages/model/src/migration.ts index 585ce88b06..0c97d1ebb0 100644 --- a/packages/model/src/migration.ts +++ b/packages/model/src/migration.ts @@ -118,6 +118,7 @@ export interface MigrationClient { wsIds: WorkspaceIds + fullReindex: () => Promise reindex: (domain: Domain, classes: Ref>[]) => Promise readonly logger: ModelLogger readonly ctx: MeasureContext diff --git a/pods/fulltext/package.json b/pods/fulltext/package.json index e9d4ac521b..267f0e3b66 100644 --- a/pods/fulltext/package.json +++ b/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" } } diff --git a/pods/fulltext/src/manager.ts b/pods/fulltext/src/manager.ts index 6e2d823d1a..e0e6c6ce0d 100644 --- a/pods/fulltext/src/manager.ts +++ b/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>( + this.txConsumer = this.opt.queue.createConsumer | TxDomainEvent>>( 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>>[], control: ConsumerControl): Promise { + private async processTransactions ( + msg: ConsumerMessage> | TxDomainEvent>>[], + control: ConsumerControl + ): Promise { 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) }) } } diff --git a/pods/fulltext/src/workspace.ts b/pods/fulltext/src/workspace.ts index 54d38e7201..e34ad95ab7 100644 --- a/pods/fulltext/src/workspace.ts +++ b/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 diff --git a/server/core/src/types.ts b/server/core/src/types.ts index d75b4ce05a..c2852e4e49 100644 --- a/server/core/src/types.ts +++ b/server/core/src/types.ts @@ -380,9 +380,16 @@ export interface FullTextAdapter { update: Record ) => Promise remove: (ctx: MeasureContext, workspace: WorkspaceUuid, id: Ref[]) => Promise + removeByQuery: (ctx: MeasureContext, workspace: WorkspaceUuid, query: DocumentQuery) => Promise clean: (ctx: MeasureContext, workspace: WorkspaceUuid) => Promise updateMany: (ctx: MeasureContext, workspace: WorkspaceUuid, docs: IndexedDoc[]) => Promise + updateByQuery: ( + ctx: MeasureContext, + workspace: WorkspaceUuid, + query: DocumentQuery, + update: Record + ) => Promise load: (ctx: MeasureContext, workspace: WorkspaceUuid, docs: Ref[]) => Promise searchString: ( ctx: MeasureContext, diff --git a/server/elastic/src/adapter.ts b/server/elastic/src/adapter.ts index 82eb7318f0..ad3117cac1 100644 --- a/server/elastic/src/adapter.ts +++ b/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>[], 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, + update: Record + ): Promise { + 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[]): Promise { 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): Promise { + 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 { 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 } } ] diff --git a/server/indexer/package.json b/server/indexer/package.json index caf65ffac2..586ff614a0 100644 --- a/server/indexer/package.json +++ b/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" } } diff --git a/server/indexer/src/indexer/indexer.ts b/server/indexer/src/indexer/indexer.ts index ebc15db068..50c5b654e2 100644 --- a/server/indexer/src/indexer/indexer.ts +++ b/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 = Omit & { date: string } + +type IndexableCommunicationEvent = + | QueueSourced + | QueueSourced + | QueueSourced + | QueueSourced + | QueueSourced + | QueueSourced + | QueueSourced + // Global Memory management configuration /** @@ -167,6 +207,8 @@ export class FullTextIndexPipeline implements FullTextPipeline { contexts: Map>, 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>[]) => 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>[] }[]> { @@ -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 { @@ -466,14 +526,193 @@ export class FullTextIndexPipeline implements FullTextPipeline { await rateLimit.waitProcessing() } - public async processDocuments (ctx: MeasureContext, result: TxCUD[], control: ConsumerControl): Promise { + async indexCommunication ( + ctx: MeasureContext, + control: ConsumerControl | undefined, + pushQueue: ElasticPushQueue + ): Promise { + const communicationApi = this.communicationApi + if (communicationApi === undefined) { + return 0 + } + let processed = 0 + const cardsInfo = new Map, _class: Ref> }>() + 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 | TxDomainEvent>)[], + control: ConsumerControl + ): Promise { const contextData = this.createContextData() ctx.contextData = contextData - // Find documents matching query + + const indexableCommunicationEventTypes: Array = [ + 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[] + const messageEvents = result.filter( + (tx) => + tx._class === core.class.TxDomainEvent && + (tx as TxDomainEvent).domain === 'communication' && + indexableCommunicationEventTypes.includes((tx as TxDomainEvent>).event.type) + ) as any as TxDomainEvent[] // 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, Ref>>(result, (it) => it.objectClass) + const byClass = groupByArray, Ref>>(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[], + toRemove: { _id: Ref, _class: Ref> }[] + ): Promise { + const communicationApi = this.communicationApi + if (communicationApi === undefined) { + return + } + const getMessage = async (cardId: CardID, msgId: MessageID): Promise => { + 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() + for (const tx of txes) { + if ( + [MessageEventType.CreateMessage, MessageEventType.UpdatePatch, MessageEventType.LinkPreviewPatch].includes( + tx.event.type as any + ) + ) { + const event = tx.event as + | QueueSourced + | QueueSourced + | QueueSourced + 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, + _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 + }) + 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>[], toRemove: { _id: Ref>, _class: Ref> }[], @@ -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, + pushQueue: ElasticPushQueue, + cardId: CardID, + cardSpace: Ref, + cardClass: Ref>, + message: Pick< + Message, + 'id' | 'edited' | 'created' | 'creator' | 'content' | 'extra' | 'blobs' | 'thread' | 'linkPreviews' + > + ): Promise { + 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, + pushQueue: ElasticPushQueue, + parentDoc: { id: Ref, _class: Ref>[], space: Ref, attachedTo?: Ref }, + blob: AttachedBlob + ): Promise { + try { + const indexedDoc: IndexedDoc = { + id: `${blob.blobId}@${parentDoc.attachedTo}` as any, + _class: [`${card.class.Card}%blob` as Ref>], + 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, + ref: Ref, + indexedDoc: IndexedDoc, + defaultContentType: string = '' + ): Promise { + 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, docInfo: Blob | undefined, indexedDoc: IndexedDoc): Promise { if (docInfo !== undefined) { const contentType = (docInfo.contentType ?? '').split(';')[0] diff --git a/server/indexer/src/indexer/utils.ts b/server/indexer/src/indexer/utils.ts index 80184db437..dc8990b88b 100644 --- a/server/indexer/src/indexer/utils.ts +++ b/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>[] | undefine } return indexedDoc } + +export const messagePseudoClass = `${cardPlugin.class.Card}%message` as Ref> +export const blobPseudoClass = `${cardPlugin.class.Card}%blob` as Ref> + +/** + * @public + */ +export function createIndexedDocFromMessage ( + cardId: Ref, + cardSpace: Ref, + cardClass: Ref>, + message: Pick +): 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 +} diff --git a/server/indexer/src/mapper.ts b/server/indexer/src/mapper.ts index b4d4d820d2..92ca02bd59 100644 --- a/server/indexer/src/mapper.ts +++ b/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 } diff --git a/server/middleware/package.json b/server/middleware/package.json index b608547d6f..60792defa3 100644 --- a/server/middleware/package.json +++ b/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" } } diff --git a/server/middleware/src/fulltext.ts b/server/middleware/src/fulltext.ts index 07cc53da2a..d69d3f12c9 100644 --- a/server/middleware/src/fulltext.ts +++ b/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>) + } } catch (err: any) { Analytics.handleError(err) } diff --git a/server/tool/src/upgrade.ts b/server/tool/src/upgrade.ts index 5926e26faf..658209e7f3 100644 --- a/server/tool/src/upgrade.ts +++ b/server/tool/src/upgrade.ts @@ -127,6 +127,10 @@ export class MigrateClientImpl implements MigrationClient { await this.lowLevel.rawDeleteMany(domain, query) } + async fullReindex (): Promise { + await this.queue.send(this.wsIds.uuid, [workspaceEvents.fullReindex()]) + } + async reindex (domain: Domain, classes: Ref>[]): Promise { await this.queue.send(this.wsIds.uuid, [workspaceEvents.reindex(domain, classes)]) } diff --git a/tests/docker-compose.yaml b/tests/docker-compose.yaml index 4579425a84..e71ea44f0f 100644 --- a/tests/docker-compose.yaml +++ b/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