execute destroyJobStream in cleanUpExportJobResources

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

@@ -39,7 +39,6 @@ class PageBulkExportJobCronService extends CronService {
       createdAt: { $lt: new Date(Date.now() - exportJobExpirationSeconds * 1000) },
     });
     for (const expiredExportJob of expiredExportJobs) {
-      pageBulkExportService?.pageBulkExportJobStreamManager?.destroyJobStream(expiredExportJob._id);
       // eslint-disable-next-line no-await-in-loop
       await this.cleanUpAndDeleteBulkExportJob(expiredExportJob);
     }

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

@@ -383,7 +383,7 @@ class PageBulkExportService {
    * - abort multipart upload
    */
   async cleanUpExportJobResources(pageBulkExportJob: PageBulkExportJobDocument) {
-    this.pageBulkExportJobStreamManager.removeJobStream(pageBulkExportJob._id);
+    this.pageBulkExportJobStreamManager?.destroyJobStream(pageBulkExportJob._id);
 
     const promises = [
       PageBulkExportPageSnapshot.deleteMany({ pageBulkExportJob }),

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

@@ -0,0 +1,38 @@
+import type { Readable } from 'stream';
+
+import type { ObjectIdLike } from '~/server/interfaces/mongoose-utils';
+
+import { BulkExportJobExpiredError, BulkExportJobRestartedError } from './errors';
+
+interface JobInProgress {
+  id: string;
+  stream: Readable | undefined;
+}
+
+class PageBulkExportJobManager {
+
+  jobsInProgress: JobInProgress[] = [];
+
+  updateJobStream(jobId: ObjectIdLike, stream: Readable): void {
+    const jobInProgress = this.jobsInProgress.find(job => job.id === jobId.toString());
+    if (jobInProgress != null) {
+      jobInProgress.stream = stream;
+    }
+  }
+
+  removeJobInProgress(jobId: ObjectIdLike, isJobRestarted = false): void {
+    const jobInProgress = this.jobsInProgress.find(job => job.id === jobId.toString());
+    if (jobInProgress == null) return;
+
+    if (jobInProgress.stream != null) {
+      if (isJobRestarted) {
+        jobInProgress.stream.destroy(new BulkExportJobRestartedError());
+      }
+      else {
+        jobInProgress.stream.destroy(new BulkExportJobExpiredError());
+      }
+    }
+    this.jobsInProgress = this.jobsInProgress.filter(job => job.id !== jobId);
+  }
+
+}