diff --git a/dev/tool/src/db.ts b/dev/tool/src/db.ts index 201489a6f9..186d7a6c20 100644 --- a/dev/tool/src/db.ts +++ b/dev/tool/src/db.ts @@ -1755,7 +1755,7 @@ export async function restoreFromv6All ( } await sendTransactorEvent(uuid, 'force-maintenance') - await restore(ctx, pipeline, wsIds, storage, { + await restore(ctx, pipeline, wsIds, storage, undefined, { date: -1, merge: false, parallel: 1, @@ -1928,7 +1928,7 @@ export async function restoreTrustedV6Workspace ( } await sendTransactorEvent(uuid, 'force-maintenance') - await restore(ctx, pipeline, wsIds, backupWsStorage, { + await restore(ctx, pipeline, wsIds, backupWsStorage, undefined, { date: -1, merge: false, parallel: 1, diff --git a/dev/tool/src/index.ts b/dev/tool/src/index.ts index e748e621ba..2636f0262d 100644 --- a/dev/tool/src/index.ts +++ b/dev/tool/src/index.ts @@ -1136,6 +1136,8 @@ export function devTool ( .option('-i, --include ', 'A list of ; separated domain names to include during backup', '*') .option('-s, --skip ', 'A list of ; separated domain names to skip during backup', '') .option('--upgrade', 'Upgrade workspace', false) + .option('--noqueue', 'NoQueue', false) + .option('--accounts', 'Restore accounts (person/socialId) from backup', false) .option( '--history-file ', 'Store blob send info into file. Will skip already send documents.', @@ -1156,6 +1158,8 @@ export function devTool ( useStorage: string historyFile: string upgrade: boolean + noqueue: boolean + accounts: boolean } ) => { await withAccountDatabase(async (db) => { @@ -1174,10 +1178,10 @@ export function devTool ( const storage = await createFileBackupStorage(dirName) const storageConfig = storageConfigFromEnv() - const queue = getPlatformQueue('tool', ws.region) - const wsProducer = queue.getProducer(toolCtx, QueueTopic.Workspace) + const queue = !cmd.noqueue ? getPlatformQueue('tool', ws.region) : undefined + const wsProducer = queue?.getProducer(toolCtx, QueueTopic.Workspace) - await wsProducer.send(toolCtx, ws.uuid, [workspaceEvents.restoring()]) + await wsProducer?.send(toolCtx, ws.uuid, [workspaceEvents.restoring()]) const workspaceStorage: StorageAdapter = buildStorageFromConfig(storageConfig) @@ -1202,7 +1206,7 @@ export function devTool ( } await sendTransactorEvent(workspace, 'force-maintenance') - await restore(toolCtx, pipeline, wsIds, storage, { + await restore(toolCtx, pipeline, wsIds, storage, cmd.accounts ? db : undefined, { date: parseInt(date ?? '-1'), merge: cmd.merge, parallel: parseInt(cmd.parallel ?? '1'), @@ -1219,12 +1223,12 @@ export function devTool ( } console.log('workspace restored') - await wsProducer.send(toolCtx, ws.uuid, [workspaceEvents.restored()]) + await wsProducer?.send(toolCtx, ws.uuid, [workspaceEvents.restored()]) } catch (err) { toolCtx.error('failed to restore', { err }) } await pipeline?.close() - await queue.shutdown() + await queue?.shutdown() await workspaceStorage?.close() }) } diff --git a/server/backup/src/backup.ts b/server/backup/src/backup.ts index 6672bc6ac4..abf9db37cd 100644 --- a/server/backup/src/backup.ts +++ b/server/backup/src/backup.ts @@ -220,7 +220,7 @@ export async function backup ( // We need to perform compaction ctx.warn('Compacting backup') await compactBackup(ctx, storage, true, { - blobLimit: options.blobDownloadLimit, + blobLimit: options.blobDownloadLimit * 1024 * 1024, skipContentTypes: options.skipBlobContentTypes, msg: { workspaceId, url: wsIds.url } }) @@ -269,12 +269,12 @@ export async function backup ( it !== DOMAIN_MODEL_TX && it !== DOMAIN_TX && it !== DOMAIN_BLOB && - it !== ('fulltext-blob' as Domain) && - !options.skipDomains.includes(it) && - (options.include === undefined || options.include.has(it)) + it !== ('fulltext-blob' as Domain) ), ...accountDomains - ] + ].filter( + (it) => !options.skipDomains.includes(it) && (options.include === undefined || options.include.has(it)) + ) ctx.info('domains for dump', { domains: domains.length, workspace: workspaceId, url: wsIds.url }) @@ -1030,7 +1030,16 @@ export async function backup ( // 1. We need to include global records based on persons/socialIdentities info which are missing in digest // 2. We need to check updates for all records present in digest const batchSize = 1000 - const toLoad = new Set([...digest.keys(), ...affectedObjects]) as Set + const toLoad = new Set( + [...digest.keys(), ...affectedObjects].filter((it) => { + try { + BigInt(it) + return true + } catch (err: any) { + return false + } + }) + ) as Set if (toLoad.size === 0) { ctx.info('No records updates') return diff --git a/server/backup/src/restore.ts b/server/backup/src/restore.ts index d441dba9c7..8454182c3e 100644 --- a/server/backup/src/restore.ts +++ b/server/backup/src/restore.ts @@ -15,7 +15,10 @@ // import { Analytics } from '@hcengineering/analytics' +import { type Person as GlobalPerson, type SocialId, type AccountDB, Account } from '@hcengineering/account' import core, { + AccountRole, + AccountUuid, Doc, Domain, DOMAIN_BLOB, @@ -37,7 +40,7 @@ import { extract } from 'tar-stream' import { createGunzip, gunzipSync } from 'zlib' import { BackupStorage } from './storage' import type { BackupInfo } from './types' -import { doTrimHash, isAccountDomain, loadDigest, migradeBlobData } from './utils' +import { chunkArray, doTrimHash, isAccountDomain, loadDigest, migradeBlobData, toAccountDomain } from './utils' export * from './storage' const dataUploadSize = 2 * 1024 * 1024 @@ -55,6 +58,7 @@ export async function restore ( pipeline: Pipeline, wsIds: WorkspaceIds, storage: BackupStorage, + accountDb: AccountDB | undefined, opt: { date: number merge?: boolean @@ -488,7 +492,88 @@ export async function restore ( } async function processAccountDomain (c: Domain): Promise { - // TODO + if (accountDb === undefined) { + ctx.info('skipping account domain restore, no accountDb provided', { domain: c }) + return + } + + const isPersonDomain = c === toAccountDomain('person') + const changeset = await loadDigest(ctx, storage, snapshots, c, opt.date) + + if (changeset.size === 0) { + ctx.info('no account domain data to restore', { domain: c }) + return + } + + ctx.info('restoring account domain', { domain: c, total: changeset.size, workspace: workspaceId }) + + const processed = new Set() + const collectedObjects: any[] = [] + + // Collect all objects from backup snapshots + for (const s of rsnapshots) { + const d = s.domains[c] + if (d === undefined) continue + + for (const sf of d.storage ?? []) { + const readStream = await storage.load(sf) + const ex = extract() + + const endPromise = new Promise((resolve, reject) => { + ex.on('entry', (headers, stream, next) => { + const name = headers.name ?? '' + if (name.endsWith('.json')) { + const objKey = name.substring(0, name.length - 5) + if (changeset.has(objKey) && !processed.has(objKey)) { + const chunks: Buffer[] = [] + stream.on('data', (chunk) => { + chunks.push(chunk) + }) + stream.on('end', () => { + try { + const obj = JSON.parse(Buffer.concat(chunks as any).toString()) + processed.add(objKey) + collectedObjects.push(obj) + } catch (err) { + ctx.warn('failed to parse account object', { name, err }) + } + next() + }) + } else { + next() + } + } else { + next() + } + stream.resume() + }) + + ex.on('finish', () => { + resolve() + }) + + const unzip = createGunzip({ level: defaultLevel }) + readStream.on('end', () => { + readStream.destroy() + }) + readStream.pipe(unzip).on('error', (err) => { + readStream.destroy() + reject(err) + }) + unzip.pipe(ex) + }) + + await endPromise + } + } + + ctx.info('collected account objects', { domain: c, count: collectedObjects.length, workspace: workspaceId }) + + if (isPersonDomain) { + await restorePersons(ctx, accountDb, collectedObjects as GlobalPerson[], wsIds) + } else { + await restoreSocialIds(ctx, accountDb, collectedObjects as SocialId[]) + } } const limiter = new RateLimiter(opt.parallel ?? 1) @@ -538,3 +623,88 @@ export async function restore ( } return true } + +const accountBatchSize = 500 + +async function restorePersons ( + ctx: MeasureContext, + accountDb: AccountDB, + persons: GlobalPerson[], + wsIds: WorkspaceIds +): Promise { + const chunks = chunkArray(persons, accountBatchSize) + for (const chunk of chunks) { + const uuids = chunk.map((p) => p.uuid) + + // Find existing persons + const existingPersons = await accountDb.person.find({ uuid: { $in: uuids } }) + const existingPersonUuids = new Set(existingPersons.map((p) => p.uuid)) + + // Insert missing persons + const personsToInsert = chunk + .filter((p) => !existingPersonUuids.has(p.uuid)) + .map((it) => { + const { '%hash%': _, ...data } = it as any + return data + }) + if (personsToInsert.length > 0) { + await accountDb.person.insertMany(personsToInsert) + ctx.info('inserted persons', { count: personsToInsert.length }) + } + + // Find existing accounts + const accountUuids = uuids as AccountUuid[] + const existingAccounts = await accountDb.account.find({ uuid: { $in: accountUuids } }) + const existingAccountUuids = new Set(existingAccounts.map((a) => a.uuid)) + + // Insert missing accounts (without password) + const accountsToInsert: Account[] = chunk + .filter((p) => !existingAccountUuids.has(p.uuid as AccountUuid)) + .map((p) => ({ uuid: p.uuid as AccountUuid })) + if (accountsToInsert.length > 0) { + await accountDb.account.insertMany(accountsToInsert) + ctx.info('inserted accounts', { count: accountsToInsert.length }) + } + + // Assign workspace to all accounts + const workspaceRoles = new Map() + for (const uuid of accountUuids) { + const role = await accountDb.getWorkspaceRole(uuid, wsIds.uuid) + if (role !== null) { + workspaceRoles.set(uuid, role) + } + } + + const toAssign: [AccountUuid, typeof wsIds.uuid, AccountRole][] = accountUuids + .filter((uuid) => !workspaceRoles.has(uuid)) + .map((uuid) => [uuid, wsIds.uuid, AccountRole.User]) + + if (toAssign.length > 0) { + await accountDb.batchAssignWorkspace(toAssign) + ctx.info('assigned workspace to accounts', { count: toAssign.length, workspace: wsIds.uuid }) + } + } +} + +async function restoreSocialIds (ctx: MeasureContext, accountDb: AccountDB, socialIds: SocialId[]): Promise { + const chunks = chunkArray(socialIds, accountBatchSize) + for (const chunk of chunks) { + const ids = chunk.map((s) => s.key) + + // Find existing socialIds + const existingSocialIds = await accountDb.socialId.find({ key: { $in: ids } }) + const existingIds = new Set(existingSocialIds.map((s) => s.key)) + + // Insert missing socialIds + const socialIdsToInsert: SocialId[] = chunk + .filter((s) => !existingIds.has(s.key)) + .map((it) => { + const { '%hash%': _1, key: _2, ...data } = it as any + return data + }) + if (socialIdsToInsert.length > 0) { + await accountDb.socialId.insertMany(socialIdsToInsert) + ctx.info('inserted socialIds', { count: socialIdsToInsert.length }) + } + } +} diff --git a/server/backup/src/service.ts b/server/backup/src/service.ts index 76887d2d94..70f5ca659c 100644 --- a/server/backup/src/service.ts +++ b/server/backup/src/service.ts @@ -28,7 +28,7 @@ import { type WorkspaceIds, type WorkspaceInfoWithStatus } from '@hcengineering/core' -import { getAccountDB } from '@hcengineering/account' +import { getAccountDB, type AccountDB } from '@hcengineering/account' import { getAccountClient } from '@hcengineering/server-client' import { type DbConfiguration, @@ -168,6 +168,9 @@ class BackupWorker { const infoTo = setInterval(() => { const avgTime = this.allBackupTime / (this.processed + 1) + if (this.activeWorkspaces.size === 0 && this.workspacesToBackup.size === 0) { + return + } ctx.warn('********** backup info **********', { processed: this.processed, toGo: this.workspacesToBackup.size, @@ -449,6 +452,8 @@ export async function doRestoreWorkspace ( pipelineFactory: PipelineFactory, skipDomains: string[], cleanIndexState: boolean, + accountsDbUrl?: string, + accountsDbNs?: string, notify?: (progress: number) => Promise ): Promise { rootCtx.warn('\nRESTORE WORKSPACE ', { @@ -457,6 +462,7 @@ export async function doRestoreWorkspace ( }) const ctx = rootCtx.newChild('doRestore', {}, { span: false }) let pipeline: Pipeline | undefined + let closeAccountDB: (() => void) | undefined try { pipeline = await pipelineFactory( ctx, @@ -472,11 +478,15 @@ export async function doRestoreWorkspace ( } const restoreIds = { uuid: bucketName as WorkspaceUuid, dataId: bucketName as WorkspaceDataId, url: '' } const storage = await createStorageBackupStorage(ctx, backupAdapter, restoreIds, wsIds.dataId ?? wsIds.uuid) + let accountDB: AccountDB | undefined + if (accountsDbUrl !== undefined) { + ;[accountDB, closeAccountDB] = await getAccountDB(accountsDbUrl, accountsDbNs) + } const result: boolean = await ctx.with( 'restore', {}, (ctx) => - restore(ctx, pipeline as Pipeline, wsIds, storage, { + restore(ctx, pipeline as Pipeline, wsIds, storage, accountDB, { date: -1, skip: new Set(skipDomains), recheck: false, // Do not need to recheck @@ -492,6 +502,7 @@ export async function doRestoreWorkspace ( rootCtx.error('\n\nFAILED to RESTORE', { workspace: wsIds.uuid, err }) return false } finally { + closeAccountDB?.() if (pipeline !== undefined) { await pipeline.close() } diff --git a/server/workspace-service/src/service.ts b/server/workspace-service/src/service.ts index 236ea991f9..bd0f7d1b86 100644 --- a/server/workspace-service/src/service.ts +++ b/server/workspace-service/src/service.ts @@ -705,6 +705,8 @@ export class WorkspaceWorker { pipelineFactory, [DOMAIN_BLOB], true, + this.accountsDbUrl, + undefined, (_p: number) => { if (progress !== Math.round(_p)) { progress = Math.round(_p)