feat: pdf export queue

apps/server/src/core/auth/dto/jwt-payload.ts

@@ -6,6 +6,7 @@ export enum JwtType {
   MFA_TOKEN = 'mfa_token',
   API_KEY = 'api_key',
   PDF_RENDER = 'pdf_render',
+  PDF_EXPORT_DOWNLOAD = 'pdf_export_download',
 }
 export type JwtPayload = {
   sub: string;
@@ -52,3 +53,9 @@ export type JwtPdfRenderPayload = {
   workspaceId: string;
   type: 'pdf_render';
 };
+
+export type JwtPdfExportDownloadPayload = {
+  fileTaskId: string;
+  workspaceId: string;
+  type: 'pdf_export_download';
+};

apps/server/src/core/auth/services/token.service.ts

@@ -13,6 +13,7 @@ import {
   JwtExchangePayload,
   JwtMfaTokenPayload,
   JwtPayload,
+  JwtPdfExportDownloadPayload,
   JwtPdfRenderPayload,
   JwtType,
 } from '../dto/jwt-payload';
@@ -128,6 +129,18 @@ export class TokenService {
     return this.jwtService.sign(payload, { expiresIn: '60s' });
   }
 
+  async generatePdfExportDownloadToken(
+    fileTaskId: string,
+    workspaceId: string,
+  ): Promise<string> {
+    const payload: JwtPdfExportDownloadPayload = {
+      fileTaskId,
+      workspaceId,
+      type: JwtType.PDF_EXPORT_DOWNLOAD,
+    };
+    return this.jwtService.sign(payload, { expiresIn: '1h' });
+  }
+
   async verifyJwt(token: string, tokenType: string) {
     const payload = await this.jwtService.verifyAsync(token, {
       secret: this.environmentService.getAppSecret(),

apps/server/src/database/migrations/20260414T124451-update-file_tasks.ts

@@ -0,0 +1,32 @@
+import { Kysely, sql } from 'kysely';
+
+export async function up(db: Kysely<any>): Promise<void> {
+  await db.schema
+    .alterTable('file_tasks')
+    .addColumn('page_id', 'uuid', (col) =>
+      col.references('pages.id').onDelete('set null').ifNotExists(),
+    )
+    .execute();
+
+  await db.schema
+    .alterTable('file_tasks')
+    .addColumn('metadata', 'jsonb', (col) => col.ifNotExists())
+    .execute();
+
+  await db.schema
+    .createIndex('idx_file_tasks_page_export')
+    .ifNotExists()
+    .on('file_tasks')
+    .columns(['page_id', 'workspace_id'])
+    .where(sql.ref('type'), '=', 'export')
+    .where(sql.ref('deleted_at'), 'is', null)
+    .execute();
+}
+
+export async function down(db: Kysely<any>): Promise<void> {
+  await db.schema.dropIndex('idx_file_tasks_page_export').execute();
+
+  await db.schema.alterTable('file_tasks').dropColumn('page_id').execute();
+
+  await db.schema.alterTable('file_tasks').dropColumn('metadata').execute();
+}

apps/server/src/database/types/db.d.ts

@@ -196,6 +196,8 @@ export interface FileTasks {
   filePath: string;
   fileSize: Int8 | null;
   id: Generated<string>;
+  metadata: Json | null;
+  pageId: string | null;
   source: string | null;
   spaceId: string | null;
   status: string | null;

apps/server/src/integrations/import/processors/file-task.processor.ts

@@ -5,6 +5,9 @@ import { QueueJob, QueueName } from 'src/integrations/queue/constants';
 import { FileImportTaskService } from '../services/file-import-task.service';
 import { FileTaskStatus } from '../utils/file.utils';
 import { StorageService } from '../../storage/storage.service';
+import { ModuleRef } from '@nestjs/core';
+import { InjectKysely } from 'nestjs-kysely';
+import { KyselyDB } from '@docmost/db/types/kysely.types';
 
 @Processor(QueueName.FILE_TASK_QUEUE)
 export class FileTaskProcessor extends WorkerHost implements OnModuleDestroy {
@@ -13,6 +16,8 @@ export class FileTaskProcessor extends WorkerHost implements OnModuleDestroy {
   constructor(
     private readonly fileTaskService: FileImportTaskService,
     private readonly storageService: StorageService,
+    private readonly moduleRef: ModuleRef,
+    @InjectKysely() private readonly db: KyselyDB,
   ) {
     super();
   }
@@ -23,8 +28,11 @@ export class FileTaskProcessor extends WorkerHost implements OnModuleDestroy {
         case QueueJob.IMPORT_TASK:
           await this.fileTaskService.processZIpImport(job.data.fileTaskId);
           break;
-        case QueueJob.EXPORT_TASK:
+        case QueueJob.PDF_EXPORT_TASK:
-          // TODO: export task
+          await this.processExportTask(job.data.fileTaskId);
+          break;
+        case QueueJob.PDF_EXPORT_CLEANUP:
+          await this.processExportCleanup();
           break;
       }
     } catch (err) {
@@ -33,6 +41,24 @@ export class FileTaskProcessor extends WorkerHost implements OnModuleDestroy {
     }
   }
 
+  private getPdfExportService() {
+    // eslint-disable-next-line @typescript-eslint/no-require-imports
+    const PdfExportModule = require('./../../../ee/pdf-export/pdf-export.service');
+    return this.moduleRef.get(PdfExportModule.PdfExportService, {
+      strict: false,
+    });
+  }
+
+  private async processExportTask(fileTaskId: string): Promise<void> {
+    const pdfExportService = this.getPdfExportService();
+    await pdfExportService.generateAndStorePdf(fileTaskId);
+  }
+
+  private async processExportCleanup(): Promise<void> {
+    const pdfExportService = this.getPdfExportService();
+    await pdfExportService.cleanupExpiredExports();
+  }
+
   @OnWorkerEvent('active')
   onActive(job: Job) {
     this.logger.debug(`Processing ${job.name} job`);
@@ -41,32 +67,39 @@ export class FileTaskProcessor extends WorkerHost implements OnModuleDestroy {
   @OnWorkerEvent('failed')
   async onFailed(job: Job) {
     this.logger.error(
-      `Error processing ${job.name} job. Import Task ID: ${job.data.fileTaskId}. Reason: ${job.failedReason}`,
+      `Error processing ${job.name} job. File Task ID: ${job.data?.fileTaskId}. Reason: ${job.failedReason}`,
     );
 
-    await this.handleFailedJob(job);
+    if (job.name === QueueJob.IMPORT_TASK) {
+      await this.handleFailedImportJob(job);
+    } else if (job.name === QueueJob.PDF_EXPORT_TASK) {
+      await this.handleFailedExportJob(job);
+    }
   }
 
   @OnWorkerEvent('completed')
   async onCompleted(job: Job) {
     this.logger.log(
-      `Completed ${job.name} job for File task ID ${job.data.fileTaskId}`,
+      `Completed ${job.name} job for File task ID ${job.data?.fileTaskId}`,
     );
 
-    try {
+    if (job.name === QueueJob.IMPORT_TASK) {
-      const fileTask = await this.fileTaskService.getFileTask(
+      try {
-        job.data.fileTaskId,
+        const fileTask = await this.fileTaskService.getFileTask(
-      );
+          job.data.fileTaskId,
-      if (fileTask) {
+        );
-        await this.storageService.delete(fileTask.filePath);
+        if (fileTask) {
-        this.logger.debug(`Deleted imported zip file: ${fileTask.filePath}`);
+          await this.storageService.delete(fileTask.filePath);
+          this.logger.debug(`Deleted imported zip file: ${fileTask.filePath}`);
+        }
+      } catch (err) {
+        this.logger.error(`Failed to delete imported zip file:`, err);
       }
-    } catch (err) {
-      this.logger.error(`Failed to delete imported zip file:`, err);
     }
+    // Export tasks: do NOT delete the file on completion (kept for 24h cache)
   }
 
-  private async handleFailedJob(job: Job) {
+  private async handleFailedImportJob(job: Job) {
     try {
       const fileTaskId = job.data.fileTaskId;
       const reason = job.failedReason || 'Unknown error';
@@ -86,6 +119,25 @@ export class FileTaskProcessor extends WorkerHost implements OnModuleDestroy {
     }
   }
 
+  private async handleFailedExportJob(job: Job) {
+    try {
+      const fileTaskId = job.data.fileTaskId;
+      const reason = job.failedReason || 'Unknown error';
+
+      await this.db
+        .updateTable('fileTasks')
+        .set({
+          status: FileTaskStatus.Failed,
+          errorMessage: reason,
+          updatedAt: new Date(),
+        })
+        .where('id', '=', fileTaskId)
+        .execute();
+    } catch (err) {
+      this.logger.error(err);
+    }
+  }
+
   async onModuleDestroy(): Promise<void> {
     if (this.worker) {
       await this.worker.close();

apps/server/src/integrations/queue/constants/queue.constants.ts

@@ -80,4 +80,7 @@ export enum QueueJob {
 
   AUDIT_LOG = 'audit-log',
   AUDIT_CLEANUP = 'audit-cleanup',
+
+  PDF_EXPORT_TASK = 'pdf-export-task',
+  PDF_EXPORT_CLEANUP = 'pdf-export-cleanup',
 }