feat: bulk page imports (#1219)

* refactor imports - WIP * Add readstream * WIP * fix attachmentId render * fix attachmentId render * turndown video tag * feat: add stream upload support and improve file handling - Add stream upload functionality to storage drivers\n- Improve ZIP file extraction with better encoding handling\n- Fix attachment ID rendering issues\n- Add AWS S3 upload stream support\n- Update dependencies for better compatibility * WIP * notion formatter * move embed parser to editor-ext package * import embeds * utility files * cleanup * Switch from happy-dom to cheerio * Refine code * WIP * bug fixes and UI * sync * WIP * sync * keep import modal mounted * Show modal during upload * WIP * WIP
2026-05-15 13:14:11 +08:00 · 2025-06-09 04:29:27 +01:00
parent ce1503af85
commit 6d024fc3de
45 changed files with 2362 additions and 149 deletions
@@ -0,0 +1,76 @@
+import { Logger, OnModuleDestroy } from '@nestjs/common';
+import { OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq';
+import { Job } from 'bullmq';
+import { QueueJob, QueueName } from 'src/integrations/queue/constants';
+import { FileTaskService } from '../services/file-task.service';
+import { FileTaskStatus } from '../utils/file.utils';
+import { StorageService } from '../../storage/storage.service';
+
+@Processor(QueueName.FILE_TASK_QUEUE)
+export class FileTaskProcessor extends WorkerHost implements OnModuleDestroy {
+  private readonly logger = new Logger(FileTaskProcessor.name);
+
+  constructor(
+    private readonly fileTaskService: FileTaskService,
+    private readonly storageService: StorageService,
+  ) {
+    super();
+  }
+
+  async process(job: Job<any, void>): Promise<void> {
+    try {
+      switch (job.name) {
+        case QueueJob.IMPORT_TASK:
+          await this.fileTaskService.processZIpImport(job.data.fileTaskId);
+          break;
+        case QueueJob.EXPORT_TASK:
+          // TODO: export task
+          break;
+      }
+    } catch (err) {
+      this.logger.error('File task failed', err);
+      throw err;
+    }
+  }
+
+  @OnWorkerEvent('active')
+  onActive(job: Job) {
+    this.logger.debug(`Processing ${job.name} job`);
+  }
+
+  @OnWorkerEvent('failed')
+  async onFailed(job: Job) {
+    this.logger.error(
+      `Error processing ${job.name} job. Reason: ${job.failedReason}`,
+    );
+
+    try {
+      const fileTaskId = job.data.fileTaskId;
+      await this.fileTaskService.updateTaskStatus(
+        fileTaskId,
+        FileTaskStatus.Failed,
+        job.failedReason,
+      );
+
+      const fileTask = await this.fileTaskService.getFileTask(fileTaskId);
+      if (fileTask) {
+        await this.storageService.delete(fileTask.filePath);
+      }
+    } catch (err) {
+      this.logger.error(err);
+    }
+  }
+
+  @OnWorkerEvent('completed')
+  onCompleted(job: Job) {
+    this.logger.log(
+      `Completed ${job.name} job for File task ID ${job.data.fileTaskId}`,
+    );
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    if (this.worker) {
+      await this.worker.close();
+    }
+  }
+}