feat: queues (#24142)

server/src/repositories/config.repository.ts

@@ -249,7 +249,7 @@ const getEnv = (): EnvData => {
         prefix: 'immich_bull',
         connection: { ...redisConfig },
         defaultJobOptions: {
-          attempts: 3,
+          attempts: 1,
           removeOnComplete: true,
           removeOnFail: false,
         },

server/src/repositories/job.repository.ts

@@ -5,11 +5,12 @@ import { JobsOptions, Queue, Worker } from 'bullmq';
 import { ClassConstructor } from 'class-transformer';
 import { setTimeout } from 'node:timers/promises';
 import { JobConfig } from 'src/decorators';
-import { JobName, JobStatus, MetadataKey, QueueCleanType, QueueName } from 'src/enum';
+import { QueueJobResponseDto, QueueJobSearchDto } from 'src/dtos/queue.dto';
+import { JobName, JobStatus, MetadataKey, QueueCleanType, QueueJobStatus, QueueName } from 'src/enum';
 import { ConfigRepository } from 'src/repositories/config.repository';
 import { EventRepository } from 'src/repositories/event.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
-import { JobCounts, JobItem, JobOf, QueueStatus } from 'src/types';
+import { JobCounts, JobItem, JobOf } from 'src/types';
 import { getKeyByValue, getMethodNames, ImmichStartupError } from 'src/utils/misc';
 
 type JobMapItem = {
@@ -115,13 +116,14 @@ export class JobRepository {
     worker.concurrency = concurrency;
   }
 
-  async getQueueStatus(name: QueueName): Promise<QueueStatus> {
+  async isActive(name: QueueName): Promise<boolean> {
     const queue = this.getQueue(name);
+    const count = await queue.getActiveCount();
+    return count > 0;
+  }
 
-    return {
-      isActive: !!(await queue.getActiveCount()),
-      isPaused: await queue.isPaused(),
-    };
+  async isPaused(name: QueueName): Promise<boolean> {
+    return this.getQueue(name).isPaused();
   }
 
   pause(name: QueueName) {
@@ -192,17 +194,28 @@ export class JobRepository {
   }
 
   async waitForQueueCompletion(...queues: QueueName[]): Promise<void> {
-    let activeQueue: QueueStatus | undefined;
-    do {
-      const statuses = await Promise.all(queues.map((name) => this.getQueueStatus(name)));
-      activeQueue = statuses.find((status) => status.isActive);
-    } while (activeQueue);
-    {
-      this.logger.verbose(`Waiting for ${activeQueue} queue to stop...`);
+    const getPending = async () => {
+      const results = await Promise.all(queues.map(async (name) => ({ pending: await this.isActive(name), name })));
+      return results.filter(({ pending }) => pending).map(({ name }) => name);
+    };
+
+    let pending = await getPending();
+
+    while (pending.length > 0) {
+      this.logger.verbose(`Waiting for ${pending[0]} queue to stop...`);
       await setTimeout(1000);
+      pending = await getPending();
     }
   }
 
+  async searchJobs(name: QueueName, dto: QueueJobSearchDto): Promise<QueueJobResponseDto[]> {
+    const jobs = await this.getQueue(name).getJobs(dto.status ?? Object.values(QueueJobStatus), 0, 1000);
+    return jobs.map((job) => {
+      const { id, name, timestamp, data } = job;
+      return { id, name: name as JobName, timestamp, data };
+    });
+  }
+
   private getJobOptions(item: JobItem): JobsOptions | null {
     switch (item.name) {
       case JobName.NotifyAlbumUpdate: {