Merge branch 'feat/78039-151411-cancel-streams-on-export-job-abort' into feat/78039-150777-page-bulk-export-job-cron-test

apps/app/src/features/page-bulk-export/server/routes/apiv3/page-bulk-export.ts

@@ -7,7 +7,8 @@ import type Crowi from '~/server/crowi';
 import type { ApiV3Response } from '~/server/routes/apiv3/interfaces/apiv3-response';
 import loggerFactory from '~/utils/logger';
 
-import { DuplicateBulkExportJobError, pageBulkExportService } from '../../service/page-bulk-export';
+import { pageBulkExportService } from '../../service/page-bulk-export';
+import { DuplicateBulkExportJobError } from '../../service/page-bulk-export/errors';
 
 const logger = loggerFactory('growi:routes:apiv3:page-bulk-export');
 

apps/app/src/features/page-bulk-export/server/service/page-bulk-export/errors.ts

@@ -0,0 +1,15 @@
+export class DuplicateBulkExportJobError extends Error {
+
+  constructor() {
+    super('Duplicate bulk export job is in progress');
+  }
+
+}
+
+export class BulkExportJobExpiredError extends Error {
+
+  constructor() {
+    super('Bulk export job has expired');
+  }
+
+}

apps/app/src/features/page-bulk-export/server/service/page-bulk-export.ts → apps/app/src/features/page-bulk-export/server/service/page-bulk-export/index.ts

@@ -1,6 +1,5 @@
 import fs from 'fs';
 import path from 'path';
-import type { Readable } from 'stream';
 import { Writable } from 'stream';
 import { pipeline as pipelinePromise } from 'stream/promises';
 
@@ -17,7 +16,6 @@ import mongoose from 'mongoose';
 import type { SupportedActionType } from '~/interfaces/activity';
 import { SupportedAction, SupportedTargetModel } from '~/interfaces/activity';
 import { AttachmentType, FilePathOnStoragePrefix } from '~/server/interfaces/attachment';
-import type { ObjectIdLike } from '~/server/interfaces/mongoose-utils';
 import type { ActivityDocument } from '~/server/models/activity';
 import { Attachment } from '~/server/models/attachment';
 import type { IAttachmentDocument } from '~/server/models/attachment';
@@ -29,11 +27,14 @@ import { preNotifyService } from '~/server/service/pre-notify';
 import { getBufferToFixedSizeTransform } from '~/server/util/stream';
 import loggerFactory from '~/utils/logger';
 
-import { PageBulkExportFormat, PageBulkExportJobInProgressStatus, PageBulkExportJobStatus } from '../../interfaces/page-bulk-export';
-import type { PageBulkExportJobDocument } from '../models/page-bulk-export-job';
-import PageBulkExportJob from '../models/page-bulk-export-job';
-import type { PageBulkExportPageSnapshotDocument } from '../models/page-bulk-export-page-snapshot';
-import PageBulkExportPageSnapshot from '../models/page-bulk-export-page-snapshot';
+import { PageBulkExportFormat, PageBulkExportJobInProgressStatus, PageBulkExportJobStatus } from '../../../interfaces/page-bulk-export';
+import type { PageBulkExportJobDocument } from '../../models/page-bulk-export-job';
+import PageBulkExportJob from '../../models/page-bulk-export-job';
+import type { PageBulkExportPageSnapshotDocument } from '../../models/page-bulk-export-page-snapshot';
+import PageBulkExportPageSnapshot from '../../models/page-bulk-export-page-snapshot';
+
+import { BulkExportJobExpiredError, DuplicateBulkExportJobError } from './errors';
+import { PageBulkExportJobStreamManager } from './page-bulk-export-job-stream-manager';
 
 
 const logger = loggerFactory('growi:services:PageBulkExportService');
@@ -43,47 +44,6 @@ type ActivityParameters ={
   endpoint: string;
 }
 
-export class DuplicateBulkExportJobError extends Error {
-
-  constructor() {
-    super('Duplicate bulk export job is in progress');
-  }
-
-}
-
-class BulkExportJobExpiredError extends Error {
-
-  constructor() {
-    super('Bulk export job has expired');
-  }
-
-}
-
-/**
- * Used to keep track of streams currently being executed, and enable destroying them
- */
-class PageBulkExportJobStreamManager {
-
-  private jobStreams: Record<string, Readable> = {};
-
-  addJobStream(jobId: ObjectIdLike, stream: Readable) {
-    this.jobStreams[jobId.toString()] = stream;
-  }
-
-  removeJobStream(jobId: ObjectIdLike) {
-    delete this.jobStreams[jobId.toString()];
-  }
-
-  destroyJobStream(jobId: ObjectIdLike) {
-    const stream = this.jobStreams[jobId.toString()];
-    if (stream != null) {
-      stream.destroy(new BulkExportJobExpiredError());
-    }
-    this.removeJobStream(jobId);
-  }
-
-}
-
 class PageBulkExportService {
 
   crowi: any;

apps/app/src/features/page-bulk-export/server/service/page-bulk-export/page-bulk-export-job-stream-manager.ts

@@ -0,0 +1,30 @@
+import type { Readable } from 'stream';
+
+import type { ObjectIdLike } from '~/server/interfaces/mongoose-utils';
+
+import { BulkExportJobExpiredError } from './errors';
+
+/**
+ * Used to keep track of streams currently being executed, and enable destroying them
+ */
+export class PageBulkExportJobStreamManager {
+
+  private jobStreams: Record<string, Readable> = {};
+
+  addJobStream(jobId: ObjectIdLike, stream: Readable): void {
+    this.jobStreams[jobId.toString()] = stream;
+  }
+
+  removeJobStream(jobId: ObjectIdLike): void {
+    delete this.jobStreams[jobId.toString()];
+  }
+
+  destroyJobStream(jobId: ObjectIdLike): void {
+    const stream = this.jobStreams[jobId.toString()];
+    if (stream != null) {
+      stream.destroy(new BulkExportJobExpiredError());
+    }
+    this.removeJobStream(jobId);
+  }
+
+}

apps/app/src/server/crowi/index.js

@@ -12,7 +12,7 @@ import pkg from '^/package.json';
 
 import { KeycloakUserGroupSyncService } from '~/features/external-user-group/server/service/keycloak-user-group-sync';
 import { LdapUserGroupSyncService } from '~/features/external-user-group/server/service/ldap-user-group-sync';
-import { PageBulkExportJobInProgressStatus, PageBulkExportJobStatus } from '~/features/page-bulk-export/interfaces/page-bulk-export';
+import { PageBulkExportJobInProgressStatus } from '~/features/page-bulk-export/interfaces/page-bulk-export';
 import PageBulkExportJob from '~/features/page-bulk-export/server/models/page-bulk-export-job';
 import instanciatePageBulkExportService, { pageBulkExportService } from '~/features/page-bulk-export/server/service/page-bulk-export';
 import instanciatePageBulkExportJobCronService, { pageBulkExportJobCronService } from '~/features/page-bulk-export/server/service/page-bulk-export-job-cron';