refactor: job events (#23161)

e2e/src/api/specs/tag.e2e-spec.ts

@@ -582,7 +582,7 @@ describe('/tags', () => {
       expect(body).toEqual([expect.objectContaining({ id: userAsset.id, success: true })]);
     });
 
-    it('should remove duplicate assets only once', async () => {
+    it.skip('should remove duplicate assets only once', async () => {
       const tagA = await create(user.accessToken, { name: 'TagA' });
       await tagAssets(
         { id: tagA.id, bulkIdsDto: { ids: [userAsset.id] } },

server/src/repositories/event.repository.ts

@@ -17,7 +17,7 @@ import { AuthDto } from 'src/dtos/auth.dto';
 import { NotificationDto } from 'src/dtos/notification.dto';
 import { ReleaseNotification, ServerVersionResponseDto } from 'src/dtos/server.dto';
 import { SyncAssetExifV1, SyncAssetV1 } from 'src/dtos/sync.dto';
-import { ImmichWorker, MetadataKey, QueueName, UserAvatarColor, UserStatus } from 'src/enum';
+import { ImmichWorker, JobStatus, MetadataKey, QueueName, UserAvatarColor, UserStatus } from 'src/enum';
 import { ConfigRepository } from 'src/repositories/config.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { JobItem, JobSource } from 'src/types';
@@ -66,8 +66,19 @@ type EventMap = {
   AssetDeleteAll: [{ assetIds: string[]; userId: string }];
   AssetRestoreAll: [{ assetIds: string[]; userId: string }];
 
+  /** a worker receives a job and emits this event to run it */
+  JobRun: [QueueName, JobItem];
+  /** job pre-hook */
   JobStart: [QueueName, JobItem];
-  JobFailed: [{ job: JobItem; error: Error | any }];
+  /** job post-hook */
+  JobComplete: [QueueName, JobItem];
+  /** job finishes without error */
+  JobSuccess: [JobSuccessEvent];
+  /** job finishes with error */
+  JobError: [JobErrorEvent];
+
+  // queue events
+  QueueStart: [QueueStartEvent];
 
   // session events
   SessionDelete: [{ sessionId: string }];
@@ -90,6 +101,13 @@ type EventMap = {
   WebsocketConnect: [{ userId: string }];
 };
 
+type JobSuccessEvent = { job: JobItem; response?: JobStatus };
+type JobErrorEvent = { job: JobItem; error: Error | any };
+
+type QueueStartEvent = {
+  name: QueueName;
+};
+
 type UserEvent = {
   name: string;
   id: string;

server/src/repositories/job.repository.ts

@@ -89,7 +89,7 @@ export class JobRepository {
       this.logger.debug(`Starting worker for queue: ${queueName}`);
       this.workers[queueName] = new Worker(
         queueName,
-        (job) => this.eventRepository.emit('JobStart', queueName, job as JobItem),
+        (job) => this.eventRepository.emit('JobRun', queueName, job as JobItem),
         { ...bull.config, concurrency: 1 },
       );
     }

server/src/services/job.service.spec.ts

@@ -222,18 +222,16 @@ describe(JobService.name, () => {
     });
   });
 
-  describe('onJobStart', () => {
+  describe('onJobRun', () => {
     it('should process a successful job', async () => {
       mocks.job.run.mockResolvedValue(JobStatus.Success);
 
-      await sut.onJobStart(QueueName.BackgroundTask, {
+      const job: JobItem = { name: JobName.FileDelete, data: { files: ['path/to/file'] } };
-        name: JobName.FileDelete,
+      await sut.onJobRun(QueueName.BackgroundTask, job);
-        data: { files: ['path/to/file'] },
-      });
 
-      expect(mocks.telemetry.jobs.addToGauge).toHaveBeenCalledWith('immich.queues.background_task.active', 1);
+      expect(mocks.event.emit).toHaveBeenCalledWith('JobStart', QueueName.BackgroundTask, job);
-      expect(mocks.telemetry.jobs.addToGauge).toHaveBeenCalledWith('immich.queues.background_task.active', -1);
+      expect(mocks.event.emit).toHaveBeenCalledWith('JobSuccess', { job, response: JobStatus.Success });
-      expect(mocks.telemetry.jobs.addToCounter).toHaveBeenCalledWith('immich.jobs.file_delete.success', 1);
+      expect(mocks.event.emit).toHaveBeenCalledWith('JobComplete', QueueName.BackgroundTask, job);
       expect(mocks.logger.error).not.toHaveBeenCalled();
     });
 
@@ -300,7 +298,7 @@ describe(JobService.name, () => {
 
         mocks.job.run.mockResolvedValue(JobStatus.Success);
 
-        await sut.onJobStart(QueueName.BackgroundTask, item);
+        await sut.onJobRun(QueueName.BackgroundTask, item);
 
         if (jobs.length > 1) {
           expect(mocks.job.queueAll).toHaveBeenCalledWith(
@@ -317,7 +315,7 @@ describe(JobService.name, () => {
       it(`should not queue any jobs when ${item.name} fails`, async () => {
         mocks.job.run.mockResolvedValue(JobStatus.Failed);
 
-        await sut.onJobStart(QueueName.BackgroundTask, item);
+        await sut.onJobRun(QueueName.BackgroundTask, item);
 
         expect(mocks.job.queueAll).not.toHaveBeenCalled();
       });

server/src/services/job.service.ts

@@ -1,6 +1,5 @@
 import { BadRequestException, Injectable } from '@nestjs/common';
 import { ClassConstructor } from 'class-transformer';
-import { snakeCase } from 'lodash';
 import { SystemConfig } from 'src/config';
 import { OnEvent } from 'src/decorators';
 import { mapAsset } from 'src/dtos/asset-response.dto';
@@ -186,7 +185,7 @@ export class JobService extends BaseService {
       throw new BadRequestException(`Job is already running`);
     }
 
-    this.telemetryRepository.jobs.addToCounter(`immich.queues.${snakeCase(name)}.started`, 1);
+    await this.eventRepository.emit('QueueStart', { name });
 
     switch (name) {
       case QueueName.VideoConversion: {
@@ -243,21 +242,19 @@ export class JobService extends BaseService {
     }
   }
 
-  @OnEvent({ name: 'JobStart' })
+  @OnEvent({ name: 'JobRun' })
-  async onJobStart(...[queueName, job]: ArgsOf<'JobStart'>) {
+  async onJobRun(...[queueName, job]: ArgsOf<'JobRun'>) {
-    const queueMetric = `immich.queues.${snakeCase(queueName)}.active`;
-    this.telemetryRepository.jobs.addToGauge(queueMetric, 1);
     try {
-      const status = await this.jobRepository.run(job);
+      await this.eventRepository.emit('JobStart', queueName, job);
-      const jobMetric = `immich.jobs.${snakeCase(job.name)}.${status}`;
+      const response = await this.jobRepository.run(job);
-      this.telemetryRepository.jobs.addToCounter(jobMetric, 1);
+      await this.eventRepository.emit('JobSuccess', { job, response });
-      if (status === JobStatus.Success || status == JobStatus.Skipped) {
+      if (response && typeof response === 'string' && [JobStatus.Success, JobStatus.Skipped].includes(response)) {
         await this.onDone(job);
       }
     } catch (error: Error | any) {
-      await this.eventRepository.emit('JobFailed', { job, error });
+      await this.eventRepository.emit('JobError', { job, error });
     } finally {
-      this.telemetryRepository.jobs.addToGauge(queueMetric, -1);
+      await this.eventRepository.emit('JobComplete', queueName, job);
     }
   }
 

server/src/services/notification.service.ts

@@ -78,8 +78,8 @@ export class NotificationService extends BaseService {
     await this.notificationRepository.cleanup();
   }
 
-  @OnEvent({ name: 'JobFailed' })
+  @OnEvent({ name: 'JobError' })
-  async onJobFailed({ job, error }: ArgOf<'JobFailed'>) {
+  async onJobError({ job, error }: ArgOf<'JobError'>) {
     const admin = await this.userRepository.getAdmin();
     if (!admin) {
       return;

server/src/services/telemetry.service.ts

@@ -1,5 +1,7 @@
+import { snakeCase } from 'lodash';
 import { OnEvent } from 'src/decorators';
-import { ImmichWorker } from 'src/enum';
+import { ImmichWorker, JobStatus } from 'src/enum';
+import { ArgOf, ArgsOf } from 'src/repositories/event.repository';
 import { BaseService } from 'src/services/base.service';
 
 export class TelemetryService extends BaseService {
@@ -23,4 +25,35 @@ export class TelemetryService extends BaseService {
   onUserRestore() {
     this.telemetryRepository.api.addToGauge(`immich.users.total`, 1);
   }
+
+  @OnEvent({ name: 'JobStart' })
+  onJobStart(...[queueName]: ArgsOf<'JobStart'>) {
+    const queueMetric = `immich.queues.${snakeCase(queueName)}.active`;
+    this.telemetryRepository.jobs.addToGauge(queueMetric, 1);
+  }
+
+  @OnEvent({ name: 'JobSuccess' })
+  onJobSuccess({ job, response }: ArgOf<'JobSuccess'>) {
+    if (response && Object.values(JobStatus).includes(response as JobStatus)) {
+      const jobMetric = `immich.jobs.${snakeCase(job.name)}.${response}`;
+      this.telemetryRepository.jobs.addToCounter(jobMetric, 1);
+    }
+  }
+
+  @OnEvent({ name: 'JobError' })
+  onJobError({ job }: ArgOf<'JobError'>) {
+    const jobMetric = `immich.jobs.${snakeCase(job.name)}.${JobStatus.Failed}`;
+    this.telemetryRepository.jobs.addToCounter(jobMetric, 1);
+  }
+
+  @OnEvent({ name: 'JobComplete' })
+  onJobComplete(...[queueName]: ArgsOf<'JobComplete'>) {
+    const queueMetric = `immich.queues.${snakeCase(queueName)}.active`;
+    this.telemetryRepository.jobs.addToGauge(queueMetric, -1);
+  }
+
+  @OnEvent({ name: 'QueueStart' })
+  onQueueStart({ name }: ArgOf<'QueueStart'>) {
+    this.telemetryRepository.jobs.addToCounter(`immich.queues.${snakeCase(name)}.started`, 1);
+  }
 }