import { type Doc, type DocumentQuery, type Domain, type FindOptions, type Hierarchy, type LowLevelStorage, type MeasureContext, MeasureMetricsContext, type ModelDb, type Ref, type WorkspaceIds, type Class } from '@hcengineering/core' import { type MigrateUpdate, type MigrationClient, type MigrationIterator, type ModelLogger } from '@hcengineering/model' import { type Pipeline, type StorageAdapter, workspaceEvents, type PlatformQueueProducer, type QueueWorkspaceMessage } from '@hcengineering/server-core' import { type AccountClient } from '@hcengineering/account-client' /** * Upgrade client implementation. */ export class MigrateClientImpl implements MigrationClient { private readonly lowLevel: LowLevelStorage readonly ctx: MeasureContext constructor ( readonly pipeline: Pipeline, readonly hierarchy: Hierarchy, readonly model: ModelDb, readonly logger: ModelLogger, readonly storageAdapter: StorageAdapter, readonly accountClient: AccountClient, readonly wsIds: WorkspaceIds, readonly queue: PlatformQueueProducer, ctx?: MeasureContext ) { if (this.pipeline.context.lowLevelStorage === undefined) { throw new Error('lowLevelStorage is not defined') } this.lowLevel = this.pipeline.context.lowLevelStorage this.ctx = ctx ?? new MeasureMetricsContext('migrateClient', {}) } migrateState = new Map>() async find( domain: Domain, query: DocumentQuery, options?: FindOptions | undefined ): Promise { return await this.lowLevel.rawFindAll(domain, query, options) } async groupBy(domain: Domain, field: string, query?: DocumentQuery

): Promise> { return await this.lowLevel.groupBy(this.ctx, domain, field, query) } async traverse( domain: Domain, query: DocumentQuery, options?: FindOptions | undefined ): Promise> { return await this.lowLevel.traverse(domain, query, options) } async update(domain: Domain, query: DocumentQuery, operations: MigrateUpdate): Promise { const t = Date.now() try { await this.lowLevel.rawUpdate(domain, query, operations) } finally { if (Date.now() - t > 1000) { this.logger.log(`update${Date.now() - t > 5000 ? 'slow' : ''}`, { domain, query, time: Date.now() - t }) } } } async bulk( domain: Domain, operations: { filter: DocumentQuery, update: MigrateUpdate }[] ): Promise { for (const ops of operations) { await this.lowLevel.rawUpdate(domain, ops.filter, ops.update) } } async move( sourceDomain: Domain, query: DocumentQuery, targetDomain: Domain, size = 500 ): Promise { const ctx = new MeasureMetricsContext('move', {}) this.logger.log('move', { sourceDomain, query }) while (true) { const source = await this.lowLevel.rawFindAll(sourceDomain, query, { limit: size }) if (source.length === 0) break await this.lowLevel.upload(ctx, targetDomain, source) await this.lowLevel.clean( ctx, sourceDomain, source.map((p) => p._id) ) } } async create(domain: Domain, doc: T | T[]): Promise { const ctx = new MeasureMetricsContext('create', {}) await this.lowLevel.upload(ctx, domain, Array.isArray(doc) ? doc : [doc]) } async delete(domain: Domain, _id: Ref): Promise { const ctx = new MeasureMetricsContext('delete', {}) await this.lowLevel.clean(ctx, domain, [_id]) } async deleteMany(domain: Domain, query: DocumentQuery): Promise { await this.lowLevel.rawDeleteMany(domain, query) } async fullReindex (): Promise { await this.queue.send(this.ctx, this.wsIds.uuid, [workspaceEvents.fullReindex()]) } async reindex (domain: Domain, classes: Ref>[]): Promise { await this.queue.send(this.ctx, this.wsIds.uuid, [workspaceEvents.reindex(domain, classes)]) } }