diff --git a/apps/silo/src/apps/notion-importer/worker/phases/base.ts b/apps/silo/src/apps/notion-importer/worker/phases/base.ts index be0d8c3106..c8d67e24f9 100644 --- a/apps/silo/src/apps/notion-importer/worker/phases/base.ts +++ b/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}`); } diff --git a/apps/silo/src/apps/notion-importer/worker/phases/phase-one.ts b/apps/silo/src/apps/notion-importer/worker/phases/phase-one.ts index ab3866bb54..7502bc75eb 100644 --- a/apps/silo/src/apps/notion-importer/worker/phases/phase-one.ts +++ b/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); } diff --git a/apps/silo/src/lib/zip-manager/extractor.ts b/apps/silo/src/lib/zip-manager/extractor.ts index a09a6bf9ba..92592c4fb5 100644 --- a/apps/silo/src/lib/zip-manager/extractor.ts +++ b/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 { + logger.info(`Extracting file from zip: ${targetFileName}`); + try { const fileSize = zipStream.size; @@ -99,6 +110,7 @@ export async function extractDirectoryFromZip( ): Promise> { // For each child node, we need to extract the content const extractedFiles = new Map(); + 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 { + logger.info(`Decompressing data with compression method: ${compressionMethod}`); // If not compressed, just return the data if (compressionMethod === 0) { return compressedData; diff --git a/apps/silo/src/lib/zip-manager/zip-manager.ts b/apps/silo/src/lib/zip-manager/zip-manager.ts index 550e5347fa..22db793277 100644 --- a/apps/silo/src/lib/zip-manager/zip-manager.ts +++ b/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);