qfix: Fix export service (#9558)

services/export/pod-export/package.json

@@ -19,6 +19,8 @@
     "_phase:docker-staging": "rushx docker:staging",
     "bundle": "node ../../../common/scripts/esbuild.js --keep-names=true",
     "docker:build": "../../../common/scripts/docker_build.sh hardcoreeng/export",
+    "docker:tbuild": "docker build -t hardcoreeng/export . --platform=linux/amd64 && ../../../common/scripts/docker_tag_push.sh hardcoreeng/export",
+    "docker:abuild": "docker build -t hardcoreeng/export . --platform=linux/arm64 && ../../../common/scripts/docker_tag_push.sh hardcoreeng/export",
     "docker:staging": "../../../common/scripts/docker_tag.sh hardcoreeng/export staging",
     "docker:push": "../../../common/scripts/docker_tag.sh hardcoreeng/export",
     "format": "format src",

services/export/pod-export/src/converter.ts

@@ -21,18 +21,26 @@ import {
   Doc,
   isId,
   MarkupBlobRef,
+  matchQuery,
   MeasureContext,
   Mixin,
   Ref,
   RefTo,
+  toIdMap,
   Type,
-  WorkspaceIds
+  WorkspaceIds,
+  type IdMap,
+  type Space
 } from '@hcengineering/core'
 import attachment from '@hcengineering/model-attachment'
 import core from '@hcengineering/model-core'
 import { StorageAdapter } from '@hcengineering/server-core'
 import { UnifiedAttachment, UnifiedDoc } from './types'
 
+interface DocCache {
+  byId: IdMap<Doc>
+  byAttached: Map<Ref<Doc>, Doc[]>
+}
 export class UnifiedConverter {
   // Fields that should not be resolved
   private readonly skipResolveFields = new Set(['_class', '_id', 'collection', 'attachedTo', 'attachedToClass'])
@@ -44,6 +52,8 @@ export class UnifiedConverter {
     private readonly wsIds: WorkspaceIds
   ) {}
 
+  documentCache = new Map<Ref<Class<Doc>>, DocCache | Promise<DocCache>>()
+
   async convert (doc: Doc, attributesOnly: boolean = false): Promise<UnifiedDoc> {
     console.log('Convert', doc._id, doc._class, (doc as any).title)
 
@@ -178,7 +188,6 @@ export class UnifiedConverter {
 
     if (type._class === core.class.RefTo) {
       const to = (type as RefTo<Doc>).to
-      console.log('RefTo', key, to, value)
       if (hierarchy.isDerived(to, core.class.Doc)) {
         refFields.push(key)
         return await this.resolveReference(value as Ref<Doc>, to)
@@ -197,14 +206,25 @@ export class UnifiedConverter {
       collectionFields.push(key)
 
       // Get all documents of the collection
-      const collectionDocs = await this.client.findAll((type as Collection<any>).of, {
-        attachedTo: docId,
-        attachedToClass: docClass,
-        collection: key
-      })
+      const { byAttached } = await this.getCache((type as Collection<any>).of)
+      const cdocs = byAttached.get(docId) ?? []
+      const collectionDocs = matchQuery(
+        cdocs,
+        {
+          attachedTo: docId,
+          attachedToClass: docClass,
+          collection: key
+        },
+        (type as Collection<any>).of,
+        hierarchy
+      )
 
       // Convert each document of the collection
-      return await Promise.all(collectionDocs.map(async (doc) => await this.convert(doc)))
+      const result: UnifiedDoc<any>[] = []
+      for (const doc of collectionDocs) {
+        result.push(await this.convert(doc, attributesOnly))
+      }
+      return result
     }
 
     if (type._class === core.class.TypeTimestamp || type._class === core.class.TypeDate) {
@@ -212,10 +232,11 @@ export class UnifiedConverter {
     }
 
     if (type._class === core.class.ArrOf) {
-      return await Promise.all(
-        (value as any[]).map(async (element) => {
-          const of = (type as ArrOf<any>).of
-          return await this.resolveAttribute(
+      const result: any[] = []
+      for (const element of value as any[]) {
+        const of = (type as ArrOf<any>).of
+        result.push(
+          await this.resolveAttribute(
             '',
             of,
             element,
@@ -227,8 +248,9 @@ export class UnifiedConverter {
             docClass,
             attributesOnly
           )
-        })
-      )
+        )
+      }
+      return result
     }
 
     return value
@@ -242,7 +264,9 @@ export class UnifiedConverter {
     if (!isId(ref)) return ref
 
     try {
-      const doc = await this.client.findOne(to, { _id: ref })
+      const { byId } = await this.getCache(to)
+
+      const doc = byId.get(ref)
       if (doc === undefined) {
         console.warn(`Referenced document not found: ${ref}`)
         return ref
@@ -256,6 +280,44 @@ export class UnifiedConverter {
     }
   }
 
+  private async getCache (to: Ref<Class<Doc<Space>>>): Promise<DocCache> {
+    let p = this.documentCache.get(to)
+    if (p instanceof Promise) {
+      p = await p
+    }
+    if (p === undefined) {
+      p = this.loadCache(to)
+      this.documentCache.set(to, p)
+      p = await p
+      this.documentCache.set(to, p)
+    }
+    return p
+  }
+
+  async loadCache (_class: Ref<Class<Doc>>): Promise<DocCache> {
+    const allIds = await this.client.findAll(_class, {}, { projection: { _id: 1 } })
+    const docs: Doc[] = []
+    console.log(`Loading cache for ${_class} with ${allIds.length} documents`)
+    while (allIds.length > 0) {
+      const batch = allIds.splice(0, 10000).map((it) => it._id)
+      const batchDocs = await this.client.findAll(_class, { _id: { $in: batch } })
+      docs.push(...batchDocs)
+    }
+
+    const byAttached = new Map<Ref<Doc>, Doc[]>()
+    for (const doc of docs) {
+      const attachedTo = (doc as any).attachedTo as Ref<Doc>
+      if (attachedTo == null) {
+        continue
+      }
+      byAttached.set(attachedTo, (byAttached.get(attachedTo) ?? []).concat(doc))
+    }
+    return {
+      byId: toIdMap(docs),
+      byAttached
+    }
+  }
+
   private async resolveMarkdown (blobRef: MarkupBlobRef): Promise<string> {
     console.log(`Resolving markup content for ${blobRef}`)
     // return 'test'
@@ -279,11 +341,18 @@ export class UnifiedConverter {
     docId: Ref<Doc>,
     docClass: Ref<Class<Doc>>
   ): Promise<UnifiedAttachment[] | undefined> {
-    const attachments = await this.client.findAll(attachment.class.Attachment, {
-      attachedTo: docId,
-      attachedToClass: docClass,
-      collection: 'attachments'
-    })
+    const { byAttached } = await this.getCache(attachment.class.Attachment)
+    const rawAttachments = byAttached.get(docId) ?? []
+    const attachments = matchQuery(
+      rawAttachments,
+      {
+        attachedTo: docId,
+        attachedToClass: docClass,
+        collection: 'attachments'
+      },
+      attachment.class.Attachment,
+      this.client.getHierarchy()
+    )
 
     if (attachments.length === 0) {
       return undefined
@@ -297,7 +366,7 @@ export class UnifiedConverter {
         size: (attachment as any).size,
         contentType: (attachment as any).contentType,
         getData: async () => {
-          const buffer = await this.storage.read(this.context, this.wsIds, attachment._id)
+          const buffer = await this.storage.read(this.context, this.wsIds, (attachment as any).file)
 
           if (buffer === undefined) {
             console.error(`Attachment not found: ${attachment._id}`)

services/export/pod-export/src/exporter.ts

@@ -13,15 +13,24 @@
 // limitations under the License.
 //
 
-import { Client, DocumentQuery, MeasureContext, WorkspaceIds } from '@hcengineering/core'
+import {
+  Client,
+  DocumentQuery,
+  MeasureContext,
+  platformNow,
+  RateLimiter,
+  toIdMap,
+  WorkspaceIds
+} from '@hcengineering/core'
 import { Class, Doc, Ref, Space } from '@hcengineering/core/types/classes'
+import { type TransformConfig } from '@hcengineering/export'
 import core from '@hcengineering/model-core'
 import { StorageAdapter } from '@hcengineering/server-core'
 import path from 'path'
 import { UnifiedConverter } from './converter'
 import { UnifiedCsvSerializer } from './csv/csv-serializer'
 import { UnifiedJsonSerializer } from './json/json-serializer'
-import { type TransformConfig } from '@hcengineering/export'
+import type { UnifiedDoc } from './types'
 
 export enum ExportFormat {
   UNIFIED = 'unified',
@@ -67,9 +76,14 @@ export class WorkspaceExporter {
       docsBySpace.get(spaceId)?.push(doc)
     }
 
+    const allSpaces = toIdMap(
+      await this.client.findAll(core.class.Space, { _id: { $in: Array.from(docsBySpace.keys()) } })
+    )
+
+    const limiter = new RateLimiter(50)
     // Process each space
     for (const [spaceId, spaceDocs] of docsBySpace) {
-      const space = await this.client.findOne(core.class.Space, { _id: spaceId })
+      const space = allSpaces.get(spaceId)
       if (space === undefined) {
         console.error(`Space not found: ${spaceId}`)
         continue
@@ -79,7 +93,24 @@ export class WorkspaceExporter {
       const spaceDir = path.join(outputDir, spaceName)
 
       // Convert all docs to UnifiedDoc format
-      const unifiedDoc = await Promise.all(spaceDocs.map((doc) => this.converter.convert(doc, attributesOnly)))
+      const unifiedDoc: UnifiedDoc<any>[] = []
+      let t = platformNow()
+      for (const sd of spaceDocs) {
+        await limiter.add(async () => {
+          unifiedDoc.push(await this.converter.convert(sd, attributesOnly))
+        })
+        const elapsed = platformNow() - t
+        if (elapsed > 2500) {
+          const memUsage = process.memoryUsage()
+          const totalMemory = memUsage.heapTotal
+          console.log(
+            `Converted ${unifiedDoc.length} documents in ${Math.round(elapsed)} ms`,
+            `Memory: ${Math.round(memUsage.heapUsed / 1024 / 1024)}MB used, ${Math.round(totalMemory / 1024 / 1024)}MB total`
+          )
+          t = platformNow()
+        }
+      }
+      await limiter.waitProcessing()
 
       if (format === ExportFormat.JSON) {
         await this.jsonSerializer.serializeSpace(unifiedDoc, outputDir, spaceName)

services/export/pod-export/src/server.ts

@@ -13,9 +13,10 @@
 // limitations under the License.
 //
 
-import { isWorkspaceLoginInfo } from '@hcengineering/account-client'
+import { getClient, isWorkspaceLoginInfo } from '@hcengineering/account-client'
 import client, { ClientSocket } from '@hcengineering/client'
 import core, {
+  AccountRole,
   AccountUuid,
   Blob,
   Class,
@@ -27,16 +28,18 @@ import core, {
   type PersonId,
   Ref,
   Space,
+  systemAccountUuid,
   TxOperations,
   WorkspaceIds
 } from '@hcengineering/core'
 import drive, { createFile, Drive } from '@hcengineering/drive'
+import exportPlugin, { type TransformConfig } from '@hcengineering/export'
 import notification from '@hcengineering/notification'
 import { setMetadata } from '@hcengineering/platform'
 import { createClient, getAccountClient, getTransactorEndpoint } from '@hcengineering/server-client'
 import { initStatisticsContext, StorageAdapter, StorageConfiguration } from '@hcengineering/server-core'
 import { buildStorageFromConfig } from '@hcengineering/server-storage'
-import { decodeToken } from '@hcengineering/server-token'
+import { decodeToken, generateToken } from '@hcengineering/server-token'
 import archiver from 'archiver'
 import cors from 'cors'
 import express, { type Express, type NextFunction, type Request, type Response } from 'express'
@@ -47,9 +50,9 @@ import { tmpdir } from 'os'
 import { basename, join } from 'path'
 import { v4 as uuid } from 'uuid'
 import WebSocket from 'ws'
+import envConfig from './config'
 import { ApiError } from './error'
 import { ExportFormat, WorkspaceExporter } from './exporter'
-import exportPlugin, { type TransformConfig } from '@hcengineering/export'
 
 const extractCookieToken = (cookie?: string): string | null => {
   if (cookie === undefined || cookie === null) {
@@ -185,8 +188,34 @@ export function createServer (storageConfig: StorageConfiguration): { app: Expre
       if (decodedToken.extra?.readonly !== undefined) {
         throw new ApiError(403, 'Forbidden')
       }
+      const isAdmin: boolean = decodedToken.extra?.admin === 'true'
 
-      const platformClient = await createPlatformClient(token)
+      const accountClient = getClient(envConfig.AccountsUrl, token)
+
+      try {
+        const info = await accountClient.getLoginWithWorkspaceInfo()
+        const winfo = info.workspaces[decodedToken.workspace]
+        if (!isAdmin) {
+          if (winfo === undefined) {
+            res.status(401).end('Invalid workspace')
+            return
+          } else {
+            if (winfo.role !== AccountRole.Owner) {
+              res.status(401).end('Not an owner of workspace')
+              return
+            }
+          }
+        }
+      } catch (err: any) {
+        res.status(401).end('Invalid workspace')
+        return
+      }
+
+      const sysToken = generateToken(systemAccountUuid, decodedToken.workspace, {
+        service: 'export'
+      })
+
+      const platformClient = await createPlatformClient(sysToken)
       const account = decodedToken.account
 
       const txOperations = new TxOperations(platformClient, socialId)
@@ -343,13 +372,13 @@ async function saveToDrive (
 
   const fileContent = await fs.readFile(archivePath)
   const blobId = uuid() as Ref<Blob>
-  await storage.put(ctx, wsIds, blobId, fileContent, 'application/gzip', fileContent.length)
+  await storage.put(ctx, wsIds, blobId, fileContent, 'application/zip', fileContent.length)
 
   await createFile(client, exportDrive, drive.ids.Root, {
     title: basename(archivePath),
     file: blobId,
     size: fileContent.length,
-    type: 'application/gzip',
+    type: 'application/zip',
     lastModified: Date.now()
   })
   return exportDrive