subscribe to IPageBulkExportJob on bulk export

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

@@ -34,8 +34,12 @@ module.exports = (crowi: Crowi): Router => {
     }
 
     const { path, format } = req.body;
+    const activityParameters = {
+      ip: req.ip,
+      endpoint: req.originalUrl,
+    };
 
-    pageBulkExportService?.bulkExportWithBasePagePath(path, req.user);
+    pageBulkExportService?.bulkExportWithBasePagePath(path, req.user, activityParameters);
     return res.apiv3({}, 204);
   });
 

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

@@ -1,7 +1,7 @@
 import type { Readable } from 'stream';
 import { Writable, pipeline } from 'stream';
 
-import { type IPage, isPopulated } from '@growi/core';
+import { type IPage, isPopulated, SubscriptionStatusType } from '@growi/core';
 import { normalizePath } from '@growi/core/dist/utils/path-utils';
 import type { Archiver } from 'archiver';
 import archiver from 'archiver';
@@ -11,7 +11,9 @@ import mongoose from 'mongoose';
 
 import { SupportedAction, SupportedTargetModel } from '~/interfaces/activity';
 import type { PageModel, PageDocument } from '~/server/models/page';
+import Subscription from '~/server/models/subscription';
 import type { IAwsMultipartUploader } from '~/server/service/file-uploader/aws/multipart-upload';
+import { preNotifyService } from '~/server/service/pre-notify';
 import { getBufferToFixedSizeTransform } from '~/server/util/stream';
 import loggerFactory from '~/utils/logger';
 
@@ -26,6 +28,11 @@ interface ArchiverWithQueue extends Archiver {
   _queue?: QueueObject<any>;
 }
 
+type ActivityParameters ={
+  ip: string;
+  endpoint: string;
+}
+
 class PageBulkExportService {
 
   crowi: any;
@@ -42,7 +49,7 @@ class PageBulkExportService {
     this.activityEvent = crowi.event('activity');
   }
 
-  async bulkExportWithBasePagePath(basePagePath: string, currentUser): Promise<void> {
+  async bulkExportWithBasePagePath(basePagePath: string, currentUser, activityParameters: ActivityParameters): Promise<void> {
     const Page = mongoose.model<IPage, PageModel>('Page');
     const basePage = await Page.findByPathAndViewer(basePagePath, currentUser, null, true);
 
@@ -58,6 +65,7 @@ class PageBulkExportService {
     const zipArchiver = this.setUpZipArchiver();
     const pagesWritable = this.getPageWritable(zipArchiver);
     const bufferToPartSizeTransform = getBufferToFixedSizeTransform(this.partSize);
+    let multipartUploadWritable: Writable;
 
     // init multipart upload
     // TODO: Create abstract interface IMultipartUploader in https://redmine.weseek.co.jp/issues/135775
@@ -67,21 +75,22 @@ class PageBulkExportService {
         throw Error('Multipart upload not available for configured file upload type');
       }
       await multipartUploader.initUpload();
+
+      const pageBulkExportJob = await PageBulkExportJob.create({
+        user: currentUser._id,
+        page: basePage._id,
+        uploadId: multipartUploader.uploadId,
+        format: PageBulkExportFormat.markdown,
+      });
+      await Subscription.upsertSubscription(currentUser, SupportedTargetModel.MODEL_PAGE_BULK_EXPORT_JOB, pageBulkExportJob, SubscriptionStatusType.SUBSCRIBE);
+
+      multipartUploadWritable = this.getMultipartUploadWritable(multipartUploader, pageBulkExportJob, currentUser, activityParameters);
     }
     catch (err) {
       await this.handleExportError(err, multipartUploader);
       return;
     }
 
-    const pageBulkExportJob = await PageBulkExportJob.create({
-      user: currentUser._id,
-      page: basePage._id,
-      uploadId: multipartUploader.uploadId,
-      format: PageBulkExportFormat.markdown,
-    });
-
-    const multipartUploadWritable = this.getMultipartUploadWritable(multipartUploader, pageBulkExportJob, currentUser);
-
     // Cannot directly pipe from pagesWritable to zipArchiver due to how the 'append' method works.
     // Hence, execution of two pipelines is required.
     pipeline(pagesReadable, pagesWritable, err => this.handleExportError(err, multipartUploader));
@@ -166,7 +175,9 @@ class PageBulkExportService {
     return zipArchiver;
   }
 
-  private getMultipartUploadWritable(multipartUploader: IAwsMultipartUploader, pageBulkExportJob: PageBulkExportJobDocument, user): Writable {
+  private getMultipartUploadWritable(
+      multipartUploader: IAwsMultipartUploader, pageBulkExportJob: PageBulkExportJobDocument, user, activityParameters: ActivityParameters,
+  ): Writable {
     let partNumber = 1;
 
     return new Writable({
@@ -191,6 +202,7 @@ class PageBulkExportService {
           await pageBulkExportJob.save();
 
           const activity = await this.crowi.activityService.createActivity({
+            ...activityParameters,
             action: SupportedAction.ACTION_PAGE_BULK_EXPORT_COMPLETED,
             user,
             targetModel: SupportedTargetModel.MODEL_PAGE_BULK_EXPORT_JOB,
@@ -199,7 +211,8 @@ class PageBulkExportService {
               username: user.username,
             },
           });
-          this.activityEvent.emit('updated', activity, page, preNotify);
+          const preNotify = preNotifyService.generatePreNotify(activity);
+          this.activityEvent.emit('updated', activity, pageBulkExportJob, preNotify);
         }
         catch (err) {
           callback(err);

apps/app/src/server/models/subscription.ts

@@ -8,7 +8,9 @@ import {
   type Types, type Document, type Model, Schema,
 } from 'mongoose';
 
-import { AllSupportedTargetModels } from '~/interfaces/activity';
+import type { IPageBulkExportJob } from '~/features/page-bulk-export/interfaces/page-bulk-export';
+import type { SupportedTargetModelType } from '~/interfaces/activity';
+import { AllSupportedTargetModels, SupportedTargetModel } from '~/interfaces/activity';
 
 import { getOrCreateModel } from '../util/mongoose-utils';
 
@@ -17,7 +19,7 @@ export interface SubscriptionDocument extends ISubscription, Document {}
 
 export interface SubscriptionModel extends Model<SubscriptionDocument> {
   findByUserIdAndTargetId(userId: Types.ObjectId | string, targetId: Types.ObjectId | string): any
-  upsertSubscription(user: Ref<IUser>, targetModel: string, target: Ref<IPage>, status: string): any
+  upsertSubscription(user: Ref<IUser>, targetModel: SupportedTargetModelType, target: Ref<IPage> | Ref<IUser> | Ref<IPageBulkExportJob>, status: string): any
   subscribeByPageId(userId: Types.ObjectId, pageId: Types.ObjectId, status: string): any
   getSubscription(target: Ref<IPage>): Promise<Ref<IUser>[]>
   getUnsubscription(target: Ref<IPage>): Promise<Ref<IUser>[]>
@@ -63,7 +65,9 @@ subscriptionSchema.statics.findByUserIdAndTargetId = function(userId, targetId)
   return this.findOne({ user: userId, target: targetId });
 };
 
-subscriptionSchema.statics.upsertSubscription = function(user, targetModel, target, status) {
+subscriptionSchema.statics.upsertSubscription = function(
+    user: Ref<IUser>, targetModel: SupportedTargetModelType, target: Ref<IPage>, status: SubscriptionStatusType,
+) {
   const query = { user, targetModel, target };
   const doc = { ...query, status };
   const options = {
@@ -73,7 +77,7 @@ subscriptionSchema.statics.upsertSubscription = function(user, targetModel, targ
 };
 
 subscriptionSchema.statics.subscribeByPageId = function(userId, pageId, status) {
-  return this.upsertSubscription(userId, 'Page', pageId, status);
+  return this.upsertSubscription(userId, SupportedTargetModel.MODEL_PAGE, pageId, status);
 };
 
 subscriptionSchema.statics.getSubscription = async function(target: Ref<IPage>) {

apps/app/src/server/service/in-app-notification.ts

@@ -51,7 +51,7 @@ export default class InAppNotificationService {
   }
 
   initActivityEventListeners(): void {
-    this.activityEvent.on('updated', async(activity: ActivityDocument, target: IUser | IPage, preNotify: PreNotify) => {
+    this.activityEvent.on('updated', async(activity: ActivityDocument, target: IUser | IPage | IPageBulkExportJob, preNotify: PreNotify) => {
       try {
         const shouldNotification = activity != null && target != null && (AllEssentialActions as ReadonlyArray<string>).includes(activity.action);
         if (shouldNotification) {