Fix backup restore (#10943)

* Fix backup clean blobs issue

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

* Restore accounts

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

* Allow skip queue for backup-restore

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

* Fix backup of wrong social ids

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

* Filter backup logs

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

* Fix doRestoreWorkspace signature after accounts-restore port

The 'Restore accounts' cherry-pick updated the workspace-service
restore call site to pass accountsDbUrl/accountsDbNs through to
doRestoreWorkspace, but doRestoreWorkspace itself was never updated
to accept them (this mismatch existed in the upstream fork too and
only surfaced once types were rebuilt). Extend doRestoreWorkspace to
open an AccountDB from the given URL/ns, mirroring the existing
doBackup pattern, and thread it into restore() so the automatic
workspace-service restore flow can also restore accounts, not just
the manual dev-tool CLI restore.

Signed-off-by: Artyom Savchenko <armisav@gmail.com>

---------

Signed-off-by: Andrey Sobolev <haiodo@gmail.com>
Signed-off-by: Artyom Savchenko <armisav@gmail.com>
Co-authored-by: Andrey Sobolev <haiodo@gmail.com>

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<PersonUuid>
+      const toLoad = new Set(
+        [...digest.keys(), ...affectedObjects].filter((it) => {
+          try {
+            BigInt(it)
+            return true
+          } catch (err: any) {
+            return false
+          }
+        })
+      ) as Set<PersonUuid>
       if (toLoad.size === 0) {
         ctx.info('No records updates')
         return

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<void> {
-    // 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<string>()
+    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<void>((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<void> {
+  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<AccountUuid, AccountRole>()
+    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<void> {
+  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 })
+    }
+  }
+}

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<void>
 ): Promise<boolean> {
   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()
     }

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)