qfix: backups using pipeline (#9396)

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

dev/tool/src/index.ts

@@ -35,10 +35,13 @@ import {
   compactBackup,
   createFileBackupStorage,
   createStorageBackupStorage,
-  restore
+  restore,
+  backupDownload
 } from '@hcengineering/server-backup'
 import serverClientPlugin, { getAccountClient, getTransactorEndpoint } from '@hcengineering/server-client'
 import {
+  createBackupPipeline,
+  createEmptyBroadcastOps,
   registerAdapterFactory,
   registerDestroyFactory,
   registerServerPlugins,
@@ -82,7 +85,6 @@ import {
   getMongoClient,
   shutdownMongo
 } from '@hcengineering/mongo'
-import { backupDownload } from '@hcengineering/server-backup/src/backup'
 
 import { getModelVersion } from '@hcengineering/model-all'
 import {
@@ -94,6 +96,7 @@ import {
 import {
   QueueTopic,
   workspaceEvents,
+  type Pipeline,
   type QueueWorkspaceMessage,
   type StorageAdapter
 } from '@hcengineering/server-core'
@@ -114,6 +117,7 @@ import { performGmailAccountMigrations } from './gmail'
 import { getToolToken, getWorkspace, getWorkspaceTransactorEndpoint } from './utils'
 
 import { createRestClient } from '@hcengineering/api-client'
+import { sendTransactorEvent } from '@hcengineering/server-tool'
 import { existsSync } from 'fs'
 import { mkdir, writeFile } from 'fs/promises'
 import { dirname } from 'path'
@@ -688,72 +692,6 @@ export function devTool (
   //                       )
   //                     }
 
-  //                     const destroyer = getWorkspaceDestroyAdapter(dbUrl)
-
-  // program
-  //   .command('restore-all')
-  //   .description('Restore workspaces to selected region DB...')
-  //   .option('-t|--timeout [timeout]', 'Timeout in days', '60')
-  //   .option('-r|--region [region]', 'Timeout in days', '')
-  //   .option('-w|--workspace [workspace]', 'Force backup of selected workspace', '')
-  //   .option('-d|--dry [dry]', 'Dry run', false)
-  //   .action(async (cmd: { timeout: string, workspace: string, region: string, dry: boolean, account: string }) => {
-  //     const { txes, dbUrl } = prepareTools()
-
-  //     const bucketName = process.env.BUCKET_NAME
-  //     if (bucketName === '' || bucketName == null) {
-  //       console.error('please provide butket name env')
-  //       process.exit(1)
-  //     }
-
-  //     const token = generateToken(systemAccountEmail, getWorkspaceId(''))
-  //     const workspaces = (await listAccountWorkspaces(token, cmd.region))
-  //       .sort((a, b) => {
-  //         const bsize = b.backupInfo?.backupSize ?? 0
-  //         const asize = a.backupInfo?.backupSize ?? 0
-  //         return bsize - asize
-  //       })
-  //       .filter((it) => cmd.workspace === '' || cmd.workspace === it.workspace)
-
-  //     for (const ws of workspaces) {
-  //       const lastVisitDays = Math.floor((Date.now() - ws.lastVisit) / 1000 / 3600 / 24)
-
-  //       toolCtx.warn('--- restoring workspace', {
-  //         url: ws.workspaceUrl,
-  //         id: ws.workspace,
-  //         lastVisitDays,
-  //         backupSize: ws.backupInfo?.blobsSize ?? 0,
-  //         mode: ws.mode
-  //       })
-  //       if (cmd.dry) {
-  //         continue
-  //       }
-  //       try {
-  //         const st = Date.now()
-  //         await backupRestore(
-  //           toolCtx,
-  //           dbUrl,
-  //           bucketName,
-  //           ws,
-  //           (dbUrl, storageAdapter) => {
-  //             const factory: PipelineFactory = createBackupPipeline(toolCtx, dbUrl, txes, {
-  //               externalStorage: storageAdapter,
-  //               usePassedCtx: true
-  //             })
-  //             return factory
-  //           },
-  //           [DOMAIN_BLOB]
-  //         )
-  //         const ed = Date.now()
-  //         toolCtx.warn('--- restoring complete', {
-  //           time: ed - st
-  //         })
-  //       } catch (err: any) {
-  //         toolCtx.error('REstore of f workspace failedarchive workspace', { workspace: ws.workspace })
-  //       }
-  //     }
-  //   })
-
   // program
   //   .command('backup-all')
   //   .description('Backup all workspaces...')
@@ -1004,6 +942,7 @@ export function devTool (
       ) => {
         const storage = await createFileBackupStorage(dirName)
         await withAccountDatabase(async (db) => {
+          const { txes, dbUrl } = prepareTools()
           const ws = await getWorkspace(db, workspace)
           if (ws === null) {
             throw new Error(`workspace ${workspace} not found`)
@@ -1013,21 +952,51 @@ export function devTool (
             dataId: ws.dataId,
             url: ws.url
           }
-          const endpoint = await getWorkspaceTransactorEndpoint(ws.uuid)
+          const storageConfig = storageConfigFromEnv()
 
-          await backup(toolCtx, endpoint, wsIds, storage, {
-            force: cmd.force,
-            include: cmd.include === '*' ? undefined : new Set(cmd.include.split(';').map((it) => it.trim())),
-            skipDomains: (cmd.skip ?? '').split(';').map((it) => it.trim()),
-            timeout: 0,
-            connectTimeout: parseInt(cmd.timeout) * 1000,
-            blobDownloadLimit: parseInt(cmd.blobLimit),
-            skipBlobContentTypes: cmd.contentTypes
-              .split(';')
-              .map((it) => it.trim())
-              .filter((it) => it.length > 0),
-            keepSnapshots: parseInt(cmd.keepSnapshots)
-          })
+          const workspaceStorage: StorageAdapter = buildStorageFromConfig(storageConfig)
+
+          let pipeline: Pipeline | undefined
+          try {
+            pipeline = await createBackupPipeline(toolCtx, dbUrl, txes, {
+              externalStorage: workspaceStorage,
+              usePassedCtx: true
+            })(
+              toolCtx,
+              {
+                uuid: ws.uuid,
+                url: ws.url ?? '',
+                dataId: ws.dataId
+              },
+              createEmptyBroadcastOps(),
+              null
+            )
+            if (pipeline === undefined) {
+              toolCtx.error('failed to restore, pipeline is undefined', { workspace })
+              return
+            }
+
+            await backup(toolCtx, pipeline, wsIds, storage, {
+              force: cmd.force,
+              include: cmd.include === '*' ? undefined : new Set(cmd.include.split(';').map((it) => it.trim())),
+              skipDomains: (cmd.skip ?? '').split(';').map((it) => it.trim()),
+              timeout: 0,
+              connectTimeout: parseInt(cmd.timeout) * 1000,
+              blobDownloadLimit: parseInt(cmd.blobLimit),
+              skipBlobContentTypes: cmd.contentTypes
+                .split(';')
+                .map((it) => it.trim())
+                .filter((it) => it.length > 0),
+              keepSnapshots: parseInt(cmd.keepSnapshots)
+            })
+          } catch (err: any) {
+            toolCtx.error('Failed to backup workspace', { err, workspace })
+          } finally {
+            if (pipeline !== undefined) {
+              await pipeline.close()
+            }
+            await workspaceStorage.close()
+          }
         })
       }
     )
@@ -1145,7 +1114,6 @@ export function devTool (
     .option('-c, --recheck', 'Force hash recheck on server', false)
     .option('-i, --include <include>', 'A list of ; separated domain names to include during backup', '*')
     .option('-s, --skip <skip>', 'A list of ; separated domain names to skip during backup', '')
-    .option('--use-storage <useStorage>', 'Use workspace storage adapter from env variable', '')
     .option('--upgrade', 'Upgrade workspace', false)
     .option(
       '--history-file <historyFile>',
@@ -1170,6 +1138,7 @@ export function devTool (
         }
       ) => {
         await withAccountDatabase(async (db) => {
+          const { txes, dbUrl } = prepareTools()
           const ws = await getWorkspace(db, workspaceId)
           if (ws === null) {
             throw new Error(`workspace ${workspaceId} not found`)
@@ -1182,32 +1151,58 @@ export function devTool (
             url: ws.url
           }
           const storage = await createFileBackupStorage(dirName)
-          const storageConfig = cmd.useStorage !== '' ? storageConfigFromEnv(process.env[cmd.useStorage]) : undefined
+          const storageConfig = storageConfigFromEnv()
 
           const queue = getPlatformQueue('tool', ws.region)
           const wsProducer = queue.getProducer<QueueWorkspaceMessage>(toolCtx, QueueTopic.Workspace)
 
           await wsProducer.send(ws.uuid, [workspaceEvents.restoring()])
 
-          const workspaceStorage: StorageAdapter | undefined =
-            storageConfig !== undefined ? buildStorageFromConfig(storageConfig) : undefined
-          await restore(toolCtx, await getWorkspaceTransactorEndpoint(workspace), wsIds, storage, {
-            date: parseInt(date ?? '-1'),
-            merge: cmd.merge,
-            parallel: parseInt(cmd.parallel ?? '1'),
-            recheck: cmd.recheck,
-            include: cmd.include === '*' ? undefined : new Set(cmd.include.split(';')),
-            skip: new Set(cmd.skip.split(';')),
-            storageAdapter: workspaceStorage,
-            historyFile: cmd.historyFile
-          })
+          const workspaceStorage: StorageAdapter = buildStorageFromConfig(storageConfig)
 
-          if (cmd.upgrade) {
-            await doUpgrade(toolCtx, workspace, true, true)
+          let pipeline: Pipeline | undefined
+          try {
+            pipeline = await createBackupPipeline(toolCtx, dbUrl, txes, {
+              externalStorage: workspaceStorage,
+              usePassedCtx: true
+            })(
+              toolCtx,
+              {
+                uuid: ws.uuid,
+                url: ws.url ?? '',
+                dataId: ws.dataId
+              },
+              createEmptyBroadcastOps(),
+              null
+            )
+            if (pipeline === undefined) {
+              toolCtx.error('failed to restore, pipeline is undefined', { workspaceId })
+              return
+            }
+            await sendTransactorEvent(workspace, 'force-maintenance')
+
+            await restore(toolCtx, pipeline, wsIds, storage, {
+              date: parseInt(date ?? '-1'),
+              merge: cmd.merge,
+              parallel: parseInt(cmd.parallel ?? '1'),
+              recheck: cmd.recheck,
+              include: cmd.include === '*' ? undefined : new Set(cmd.include.split(';')),
+              skip: new Set(cmd.skip.split(';')),
+              historyFile: cmd.historyFile
+            })
+
+            if (cmd.upgrade) {
+              await doUpgrade(toolCtx, workspace, true, true)
+            } else {
+              await sendTransactorEvent(workspace, 'force-close')
+            }
+
+            console.log('workspace restored')
+            await wsProducer.send(ws.uuid, [workspaceEvents.restored()])
+          } catch (err) {
+            toolCtx.error('failed to restore', { err })
           }
-
-          console.log('workspace restored')
-          await wsProducer.send(ws.uuid, [workspaceEvents.restored()])
+          await pipeline?.close()
           await queue.shutdown()
           await workspaceStorage?.close()
         })

dev/tool/src/utils.ts

@@ -1,7 +1,5 @@
 import { getWorkspaceById, getWorkspaceByUrl, type AccountDB, type Workspace } from '@hcengineering/account'
 import {
-  systemAccountUuid,
-  type WorkspaceUuid,
   type AttachedData,
   type AttachedDoc,
   type Class,
@@ -9,10 +7,9 @@ import {
   type DocumentUpdate,
   type Ref,
   type Space,
-  type TxOperations
+  type TxOperations,
+  type WorkspaceUuid
 } from '@hcengineering/core'
-import { getTransactorEndpoint } from '@hcengineering/server-client'
-import { generateToken } from '@hcengineering/server-token'
 
 export async function findOrUpdateAttached<T extends AttachedDoc> (
   client: TxOperations,
@@ -61,13 +58,4 @@ export async function getWorkspace (db: AccountDB, workspace: string): Promise<W
   return wsObj
 }
 
-export function getToolToken (workspace?: WorkspaceUuid): string {
-  return generateToken(systemAccountUuid, workspace, { service: 'tool' })
-}
-
-export async function getWorkspaceTransactorEndpoint (
-  workspace: WorkspaceUuid,
-  type: 'external' | 'internal' = 'external'
-): Promise<string> {
-  return await getTransactorEndpoint(getToolToken(workspace), type)
-}
+export { getToolToken, getWorkspaceTransactorEndpoint } from '@hcengineering/server-tool'

dev/tool/src/workspace.ts

@@ -20,26 +20,13 @@ import core, {
   type Client as CoreClient,
   type Doc,
   DOMAIN_TX,
-  type MeasureContext,
   type Ref,
   type Tx,
-  type WorkspaceDataId,
-  type WorkspaceIds,
-  type WorkspaceInfoWithStatus,
   type WorkspaceUuid
 } from '@hcengineering/core'
 import { getMongoClient, getWorkspaceMongoDB } from '@hcengineering/mongo'
-import { createStorageBackupStorage, restore } from '@hcengineering/server-backup'
-import {
-  createDummyStorageAdapter,
-  type PipelineFactory,
-  type StorageAdapter,
-  wrapPipeline
-} from '@hcengineering/server-core'
-import { createStorageFromConfig, storageConfigFromEnv } from '@hcengineering/server-storage'
 import { connect } from '@hcengineering/server-tool'
 import { generateModelDiff, printDiff } from './mdiff'
-import { createEmptyBroadcastOps } from '@hcengineering/server-pipeline'
 
 export async function diffWorkspace (mongoUrl: string, dbName: string, rawTxes: Tx[]): Promise<void> {
   const client = getMongoClient(mongoUrl)
@@ -117,60 +104,3 @@ export async function updateField (
     await connection.close()
   }
 }
-
-export async function backupRestore (
-  ctx: MeasureContext,
-  dbURL: string,
-  bucketName: string,
-  workspace: WorkspaceInfoWithStatus,
-  pipelineFactoryFactory: (mongoUrl: string, storage: StorageAdapter) => PipelineFactory,
-  skipDomains: string[]
-): Promise<boolean> {
-  const storageEnv = process.env.STORAGE
-  if (storageEnv === undefined) {
-    console.error('please provide STORAGE env')
-    process.exit(1)
-  }
-  if (bucketName.trim() === '') {
-    console.error('please provide butket name env')
-    process.exit(1)
-  }
-  const backupStorageConfig = storageConfigFromEnv(storageEnv)
-
-  const storageAdapter = createStorageFromConfig(backupStorageConfig.storages[0])
-
-  const workspaceStorage = createDummyStorageAdapter()
-  const pipelineFactory = pipelineFactoryFactory(dbURL, workspaceStorage)
-
-  try {
-    const storage = await createStorageBackupStorage(
-      ctx,
-      storageAdapter,
-      {
-        uuid: 'backup' as WorkspaceUuid,
-        url: bucketName,
-        dataId: bucketName as WorkspaceDataId
-      },
-      workspace.dataId ?? workspace.uuid
-    )
-    const wsUrl: WorkspaceIds = {
-      uuid: workspace.uuid,
-      dataId: workspace.dataId,
-      url: workspace.url
-    }
-    const result: boolean = await ctx.with('restore', {}, (ctx) =>
-      restore(ctx, '', wsUrl, storage, {
-        date: -1,
-        skip: new Set(skipDomains),
-        recheck: false,
-        storageAdapter: workspaceStorage,
-        getConnection: async () => {
-          return wrapPipeline(ctx, await pipelineFactory(ctx, wsUrl, createEmptyBroadcastOps(), null), wsUrl)
-        }
-      })
-    )
-    return result
-  } finally {
-    await storageAdapter.close()
-  }
-}