Merge pull request #10486 from growilabs/feat/94790-172032-add-audit-log-bulk-export-job-cron

feat: add audit log bulk export job cron

apps/app/src/features/audit-log-bulk-export/server/service/audit-log-bulk-export-job-cron/index.ts

@@ -0,0 +1,167 @@
+import type { IUser } from '@growi/core';
+import { getIdForRef, isPopulated } from '@growi/core';
+import mongoose from 'mongoose';
+
+import type { SupportedActionType } from '~/interfaces/activity';
+import { SupportedAction, SupportedTargetModel } from '~/interfaces/activity';
+import type Crowi from '~/server/crowi';
+import CronService from '~/server/service/cron';
+import loggerFactory from '~/utils/logger';
+
+import {
+  AuditLogBulkExportJobInProgressJobStatus,
+  AuditLogBulkExportJobStatus,
+} from '../../../interfaces/audit-log-bulk-export';
+import type { AuditLogBulkExportJobDocument } from '../../models/audit-log-bulk-export-job';
+import AuditLogBulkExportJob from '../../models/audit-log-bulk-export-job';
+
+const logger = loggerFactory('growi:service:audit-log-export-job-cron');
+
+export interface IAuditLogBulkExportJobCronService {
+  crowi: Crowi;
+  proceedBulkExportJob(
+    auditLogBulkExportJob: AuditLogBulkExportJobDocument,
+  ): void;
+  notifyExportResultAndCleanUp(
+    action: SupportedActionType,
+    auditLogBulkExportJob: AuditLogBulkExportJobDocument,
+  ): Promise<void>;
+}
+
+import type { ActivityDocument } from '~/server/models/activity';
+import { preNotifyService } from '~/server/service/pre-notify';
+import { compressAndUpload } from './steps/compress-and-upload';
+import { exportAuditLogsToFsAsync } from './steps/exportAuditLogsToFsAsync';
+
+/**
+ * Manages cronjob which proceeds AuditLogBulkExportJobs in progress.
+ * If AuditLogBulkExportJob finishes the current step, the next step will be started on the next cron execution.
+ */
+class AuditLogBulkExportJobCronService
+  extends CronService
+  implements IAuditLogBulkExportJobCronService
+{
+  crowi: Crowi;
+
+  activityEvent: NodeJS.EventEmitter;
+
+  private parallelExecLimit: number;
+
+  constructor(crowi: Crowi) {
+    super();
+    this.crowi = crowi;
+    this.activityEvent = crowi.event('activity');
+    this.parallelExecLimit = 1;
+  }
+
+  override getCronSchedule(): string {
+    return '*/10 * * * * *';
+  }
+
+  override async executeJob(): Promise<void> {
+    const auditLogBulkExportJobInProgress = await AuditLogBulkExportJob.find({
+      $or: Object.values(AuditLogBulkExportJobInProgressJobStatus).map(
+        (status) => ({
+          status,
+        }),
+      ),
+    })
+      .sort({ createdAt: 1 })
+      .limit(this.parallelExecLimit);
+    await Promise.all(
+      auditLogBulkExportJobInProgress.map((job) =>
+        this.proceedBulkExportJob(job),
+      ),
+    );
+  }
+
+  async proceedBulkExportJob(
+    auditLogBulkExportJob: AuditLogBulkExportJobDocument,
+  ) {
+    try {
+      const User = mongoose.model<IUser>('User');
+      const user = await User.findById(getIdForRef(auditLogBulkExportJob.user));
+
+      if (!user) {
+        throw new Error(
+          `User not found for audit log export job: ${auditLogBulkExportJob._id}`,
+        );
+      }
+
+      if (
+        auditLogBulkExportJob.status === AuditLogBulkExportJobStatus.exporting
+      ) {
+        await exportAuditLogsToFsAsync.bind(this)(auditLogBulkExportJob);
+      } else if (
+        auditLogBulkExportJob.status === AuditLogBulkExportJobStatus.uploading
+      ) {
+        await compressAndUpload.bind(this)(auditLogBulkExportJob);
+      }
+    } catch (err) {
+      logger.error(err);
+    }
+  }
+
+  async notifyExportResultAndCleanUp(
+    action: SupportedActionType,
+    auditLogBulkExportJob: AuditLogBulkExportJobDocument,
+  ): Promise<void> {
+    auditLogBulkExportJob.status =
+      action === SupportedAction.ACTION_AUDIT_LOG_BULK_EXPORT_COMPLETED
+        ? AuditLogBulkExportJobStatus.completed
+        : AuditLogBulkExportJobStatus.failed;
+
+    try {
+      await auditLogBulkExportJob.save();
+      await this.notifyExportResult(auditLogBulkExportJob, action);
+    } catch (err) {
+      logger.error(err);
+    }
+    // TODO: Implement cleanup process in a future task.
+    // The following method `cleanUpExportJobResources` will be called here once it's ready.
+  }
+
+  private async notifyExportResult(
+    auditLogBulkExportJob: AuditLogBulkExportJobDocument,
+    action: SupportedActionType,
+  ) {
+    logger.debug(
+      'Creating activity with targetModel:',
+      SupportedTargetModel.MODEL_AUDIT_LOG_BULK_EXPORT_JOB,
+    );
+    const activity = await this.crowi.activityService.createActivity({
+      action,
+      targetModel: SupportedTargetModel.MODEL_AUDIT_LOG_BULK_EXPORT_JOB,
+      target: auditLogBulkExportJob,
+      user: auditLogBulkExportJob.user,
+      snapshot: {
+        username: isPopulated(auditLogBulkExportJob.user)
+          ? auditLogBulkExportJob.user.username
+          : '',
+      },
+    });
+    const getAdditionalTargetUsers = async (activity: ActivityDocument) => [
+      activity.user,
+    ];
+    const preNotify = preNotifyService.generatePreNotify(
+      activity,
+      getAdditionalTargetUsers,
+    );
+    this.activityEvent.emit(
+      'updated',
+      activity,
+      auditLogBulkExportJob,
+      preNotify,
+    );
+  }
+}
+
+// eslint-disable-next-line import/no-mutable-exports
+export let auditLogBulkExportJobCronService:
+  | AuditLogBulkExportJobCronService
+  | undefined;
+export default function instantiate(crowi: Crowi): void {
+  auditLogBulkExportJobCronService = new AuditLogBulkExportJobCronService(
+    crowi,
+  );
+}

apps/app/src/features/audit-log-bulk-export/server/service/audit-log-bulk-export-job-cron/steps/compress-and-upload.ts

@@ -0,0 +1,19 @@
+import { SupportedAction } from '~/interfaces/activity';
+import type { AuditLogBulkExportJobDocument } from '../../../models/audit-log-bulk-export-job';
+import type { IAuditLogBulkExportJobCronService } from '..';
+/**
+ * Execute a pipeline that reads the audit log files from the temporal fs directory,
+ * compresses them into a zip file, and uploads to the cloud storage.
+ *
+ * TODO: Implement the actual compression and upload logic in a future task.
+ * Currently, this function only notifies a successful export completion.
+ */
+export async function compressAndUpload(
+  this: IAuditLogBulkExportJobCronService,
+  job: AuditLogBulkExportJobDocument,
+): Promise<void> {
+  await this.notifyExportResultAndCleanUp(
+    SupportedAction.ACTION_AUDIT_LOG_BULK_EXPORT_COMPLETED,
+    job,
+  );
+}

apps/app/src/features/audit-log-bulk-export/server/service/audit-log-bulk-export-job-cron/steps/exportAuditLogsToFsAsync.ts

@@ -0,0 +1,17 @@
+import { AuditLogBulkExportJobStatus } from '~/features/audit-log-bulk-export/interfaces/audit-log-bulk-export';
+import type { AuditLogBulkExportJobDocument } from '../../../models/audit-log-bulk-export-job';
+import type { IAuditLogBulkExportJobCronService } from '..';
+
+/**
+ * Export audit logs to the file system before compressing and uploading.
+ *
+ * TODO: Implement the actual export logic in a later task.
+ * For now, this function only updates the job status to `uploading`.
+ */
+export async function exportAuditLogsToFsAsync(
+  this: IAuditLogBulkExportJobCronService,
+  job: AuditLogBulkExportJobDocument,
+): Promise<void> {
+  job.status = AuditLogBulkExportJobStatus.uploading;
+  await job.save();
+}

apps/app/src/features/audit-log-bulk-export/server/service/check-audit-log-bulk-export-job-in-progress-cron.ts

@@ -3,6 +3,7 @@ import CronService from '~/server/service/cron';
 
 import { AuditLogBulkExportJobInProgressJobStatus } from '../../interfaces/audit-log-bulk-export';
 import AuditLogExportJob from '../models/audit-log-bulk-export-job';
+import { auditLogBulkExportJobCronService } from './audit-log-bulk-export-job-cron';
 
 /**
  * Manages cronjob which checks if AuditLogExportJob in progress exists.
@@ -26,9 +27,13 @@ class CheckAuditLogBulkExportJobInProgressCronService extends CronService {
     });
     const auditLogExportInProgressExists = auditLogExportJobInProgress != null;
 
-    if (auditLogExportInProgressExists) {
-      // TODO: Start the cron that actually performs audit-log export.
-      // This will be implemented in a later task.
+    if (
+      auditLogExportInProgressExists &&
+      !auditLogBulkExportJobCronService?.isJobRunning()
+    ) {
+      auditLogBulkExportJobCronService?.startCron();
+    } else if (!auditLogExportInProgressExists) {
+      auditLogBulkExportJobCronService?.stopCron();
     }
   }
 }

apps/app/src/interfaces/activity.ts

@@ -5,6 +5,7 @@ const MODEL_PAGE = 'Page';
 const MODEL_USER = 'User';
 const MODEL_COMMENT = 'Comment';
 const MODEL_PAGE_BULK_EXPORT_JOB = 'PageBulkExportJob';
+const MODEL_AUDIT_LOG_BULK_EXPORT_JOB = 'AuditLogBulkExportJob';
 
 // Action
 const ACTION_UNSETTLED = 'UNSETTLED';
@@ -59,6 +60,11 @@ const ACTION_PAGE_EXPORT = 'PAGE_EXPORT';
 const ACTION_PAGE_BULK_EXPORT_COMPLETED = 'PAGE_BULK_EXPORT_COMPLETED';
 const ACTION_PAGE_BULK_EXPORT_FAILED = 'PAGE_BULK_EXPORT_FAILED';
 const ACTION_PAGE_BULK_EXPORT_JOB_EXPIRED = 'PAGE_BULK_EXPORT_JOB_EXPIRED';
+const ACTION_AUDIT_LOG_BULK_EXPORT_COMPLETED =
+  'AUDIT_LOG_BULK_EXPORT_COMPLETED';
+const ACTION_AUDIT_LOG_BULK_EXPORT_FAILED = 'AUDIT_LOG_BULK_EXPORT_FAILED';
+const ACTION_AUDIT_LOG_BULK_EXPORT_JOB_EXPIRED =
+  'AUDIT_LOG_BULK_EXPORT_JOB_EXPIRED';
 const ACTION_TAG_UPDATE = 'TAG_UPDATE';
 const ACTION_IN_APP_NOTIFICATION_ALL_STATUSES_OPEN =
   'IN_APP_NOTIFICATION_ALL_STATUSES_OPEN';
@@ -195,6 +201,7 @@ export const SupportedTargetModel = {
   MODEL_PAGE,
   MODEL_USER,
   MODEL_PAGE_BULK_EXPORT_JOB,
+  MODEL_AUDIT_LOG_BULK_EXPORT_JOB,
 } as const;
 
 export const SupportedEventModel = {
@@ -373,6 +380,9 @@ export const SupportedAction = {
   ACTION_PAGE_BULK_EXPORT_COMPLETED,
   ACTION_PAGE_BULK_EXPORT_FAILED,
   ACTION_PAGE_BULK_EXPORT_JOB_EXPIRED,
+  ACTION_AUDIT_LOG_BULK_EXPORT_COMPLETED,
+  ACTION_AUDIT_LOG_BULK_EXPORT_FAILED,
+  ACTION_AUDIT_LOG_BULK_EXPORT_JOB_EXPIRED,
 } as const;
 
 // Action required for notification
@@ -394,6 +404,9 @@ export const EssentialActionGroup = {
   ACTION_PAGE_BULK_EXPORT_COMPLETED,
   ACTION_PAGE_BULK_EXPORT_FAILED,
   ACTION_PAGE_BULK_EXPORT_JOB_EXPIRED,
+  ACTION_AUDIT_LOG_BULK_EXPORT_COMPLETED,
+  ACTION_AUDIT_LOG_BULK_EXPORT_FAILED,
+  ACTION_AUDIT_LOG_BULK_EXPORT_JOB_EXPIRED,
 } as const;
 
 export const ActionGroupSize = {

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

@@ -8,6 +8,7 @@ import lsxRoutes from '@growi/remark-lsx/dist/server/index.cjs';
 import mongoose from 'mongoose';
 import next from 'next';
 
+import instantiateAuditLogBulkExportJobCronService from '~/features/audit-log-bulk-export/server/service/audit-log-bulk-export-job-cron';
 import { checkAuditLogExportJobInProgressCronService } from '~/features/audit-log-bulk-export/server/service/check-audit-log-bulk-export-job-in-progress-cron';
 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';
@@ -365,6 +366,7 @@ Crowi.prototype.setupCron = function() {
   instanciatePageBulkExportJobCleanUpCronService(this);
   pageBulkExportJobCleanUpCronService.startCron();
 
+  instantiateAuditLogBulkExportJobCronService(this);
   checkAuditLogExportJobInProgressCronService.startCron();
 
   startOpenaiCronIfEnabled();