[SILO-445] fix: sequentially upload assets in notion import (#3902)

- Limit max attachment size to 50MB

apps/silo/src/apps/notion-importer/worker/phases/base.ts

@@ -134,6 +134,7 @@ export abstract class NotionMigratorBase extends TaskHandler {
       if (!currentNode) {
         // Get the file tree and initialize counter
         currentNode = await zipDriver.buildFileTree();
+        logger.info(`After building file tree, current node: ${currentNode?.name}`);
         if (!currentNode) {
           throw new Error(`Current node not found for job ${jobId}`);
         }

apps/silo/src/apps/notion-importer/worker/phases/phase-one.ts

@@ -1,8 +1,13 @@
-import * as bluebird from "bluebird";
 import mimetics from "mimetics";
 import { Client, ExPage } from "@plane/sdk";
 import { TImportJob, TPage } from "@plane/types";
-import { ENotionImporterKeyType, ENotionMigrationType, TAssetInfo, TNotionMigratorData } from "@/apps/notion-importer/types";
+import {
+  ENotionImporterKeyType,
+  ENotionMigrationType,
+  TAssetInfo,
+  TNotionMigratorData,
+} from "@/apps/notion-importer/types";
+import { protect } from "@/lib/errors";
 import { TZipFileNode } from "@/lib/zip-manager";
 import { logger } from "@/logger";
 import { getAPIClient } from "@/services/client";
@@ -163,49 +168,46 @@ export class NotionPhaseOneMigrator extends NotionMigratorBase {
      */
     const parentPath = root.path.split("/").pop();
 
-    await bluebird.Promise.map(
-      attachmentNodes,
-      async (node) => {
-        // Check if the content is present in the content map
-        const content = contentMap.get(node.path);
-        if (!content) {
-          logger.error(`Content not found for attachment node ${node.path}`, {
-            jobId: job.id,
-          });
-          return;
-        }
+    for (const node of attachmentNodes) {
+      // Check if the content is present in the content map
+      const content = contentMap.get(node.path);
+      // skip processing if content is empty (Buffer.alloc(0)) or undefined
+      if (!content || content.length === 0) {
+        logger.error(`Content not found for attachment node ${node.path}`, {
+          jobId: job.id,
+        });
+        continue;
+      }
 
-        try {
-          // Parse the content to get the mime type
-          const parsed = mimetics.parse(content);
-          // Upload the asset and get the asset id
-          const assetId = await client.assets.uploadAsset(
-            job.workspace_slug,
-            new File([content], node.name, {
-              type: parsed?.mime,
-            }),
-            node.name,
-            content.length
-          );
+      try {
+        // Parse the content to get the mime type
+        const parsed = mimetics.parse(content);
+        // Upload the asset and get the asset id
+        const assetId = await protect(
+          client.assets.uploadAsset,
+          job.workspace_slug,
+          new File([content], node.name, {
+            type: parsed?.mime,
+          }),
+          node.name,
+          content.length
+        );
 
-          const assetInfo: TAssetInfo = {
-            id: assetId,
-            name: node.name,
-            type: parsed?.mime ?? "application/octet-stream",
-            size: content.length,
-          }
+        const assetInfo: TAssetInfo = {
+          id: assetId,
+          name: node.name,
+          type: parsed?.mime ?? "application/octet-stream",
+          size: content.length,
+        };
 
-          result.set(`${parentPath}/${node.name}`, JSON.stringify(assetInfo));
-
-        } catch (error) {
-          logger.error(`Error uploading asset for job ${job.id} and node ${node.name}`, {
-            jobId: job.id,
-            error,
-          }); // TODO: What do we do with this error?
-        }
-      },
-      { concurrency: 3 }
-    );
+        result.set(`${parentPath}/${node.name}`, JSON.stringify(assetInfo));
+      } catch (error) {
+        logger.error(`Error uploading asset for job ${job.id} and node ${node.name}`, {
+          jobId: job.id,
+          error,
+        }); // TODO: What do we do with this error?
+      }
+    }
 
     if (result.size > 0) await this.setCacheObjects(fileId, ENotionImporterKeyType.ASSET, result);
   }

apps/silo/src/lib/zip-manager/extractor.ts

@@ -3,6 +3,15 @@ import { logger } from "@/logger";
 import { TZipFileNode, EZipNodeType } from "./types";
 import { ZipStream } from "./zip-stream";
 
+// ======================= CONSTANTS ======================= //
+/**
+ * The maximum file size for an attachment in bytes
+ * This is a temporary solution to avoid extracting too large files
+ * and causing the worker to crash
+ * Files more than 50MB will be skipped
+ */
+const ATTACHMENT_MAX_FILE_SIZE = 50 * 1024 * 1024; // 50MB
+
 // ======================= PUBLIC API FUNCTIONS ======================= //
 
 /**
@@ -58,6 +67,8 @@ export async function extractZipTableOfContents(zipStream: ZipStream): Promise<s
  * @returns Promise resolving to a Buffer containing the file contents
  */
 export async function extractFileFromZip(zipStream: ZipStream, targetFileName: string): Promise<Buffer> {
+  logger.info(`Extracting file from zip: ${targetFileName}`);
+
   try {
     const fileSize = zipStream.size;
 
@@ -99,6 +110,7 @@ export async function extractDirectoryFromZip(
 ): Promise<Map<string, Buffer>> {
   // For each child node, we need to extract the content
   const extractedFiles = new Map<string, Buffer>();
+  logger.info(`Extracting directory from zip: ${directoryNode.path}`);
 
   if (!directoryNode.children) {
     return extractedFiles;
@@ -390,6 +402,7 @@ interface ZipFileEntry {
   localHeaderOffset: number;
   compressedSize: number;
   compressionMethod: number;
+  fileName: string;
 }
 
 /**
@@ -435,6 +448,7 @@ function findFileEntryInCentralDirectory(
         localHeaderOffset,
         compressedSize,
         compressionMethod,
+        fileName,
       };
     }
 
@@ -469,6 +483,17 @@ async function extractFileData(zipStream: ZipStream, fileEntry: ZipFileEntry): P
   // File data starts after the header, filename, and extra field
   const fileDataOffset = fileEntry.localHeaderOffset + 30 + localFileNameLength + localExtraFieldLength;
 
+  // Get the file size of the particular file
+  const fileSize = fileEntry.compressedSize;
+
+  if (fileSize > ATTACHMENT_MAX_FILE_SIZE) {
+    logger.info(`File size ${fileEntry.fileName} is greater than max allowed size, skipping extraction`, {
+      fileSize,
+      fileEntry,
+    });
+    return Buffer.alloc(0);
+  }
+
   // Read the file data
   await zipStream.seek(fileDataOffset);
   const compressedData = await zipStream.read(fileEntry.compressedSize);
@@ -485,6 +510,7 @@ async function extractFileData(zipStream: ZipStream, fileEntry: ZipFileEntry): P
  * @returns Promise resolving to the decompressed data
  */
 async function decompressData(compressedData: Buffer, compressionMethod: number): Promise<Buffer> {
+  logger.info(`Decompressing data with compression method: ${compressionMethod}`);
   // If not compressed, just return the data
   if (compressionMethod === 0) {
     return compressedData;

apps/silo/src/lib/zip-manager/zip-manager.ts

@@ -1,3 +1,4 @@
+import { logger } from "@/logger";
 import { Store } from "@/worker/base";
 import { extractZipTableOfContents, extractDirectoryFromZip, extractFileFromZip } from "./extractor";
 import { StorageProvider } from "./storage-provider";
@@ -35,7 +36,7 @@ export class ZipManager {
     this.zipStream = new ZipStream(this.storageProvider, fileId, contentLength);
 
     // Table of contents is a list of the file paths which are present in the zip file
-    this.toc = await this.getTableOfContents();
+    await this.getTableOfContents();
   }
 
   /**
@@ -51,6 +52,10 @@ export class ZipManager {
       throw new Error("Zip stream not initialized");
     }
 
+    if (this.toc.length > 0) {
+      return this.toc;
+    }
+
     // Let's check if our cache has the TOC
     const cachedTOC = await this.store.get(getKey(this.fileId, TOC_KEY));
     if (cachedTOC) {
@@ -59,6 +64,9 @@ export class ZipManager {
 
     // If not, let's fetch it from the zip stream
     this.toc = await extractZipTableOfContents(this.zipStream);
+
+    logger.info(`Fetched TOC for file ${this.fileId}`, { toc: this.toc });
+
     const ttl = 60 * 60 * 4; // 4 hours
     await this.store.set(getKey(this.fileId, TOC_KEY), JSON.stringify(this.toc), ttl);