refactor(server): flatten infra folders (#2120)

* refactor: flatten infra folders * fix: database migrations * fix: test related import * fix: github actions workflow * chore: rename schemas to typesense-schemas
2025-12-08 20:29:05 +00:00 · 2023-03-30 15:38:55 -04:00
parent 468e620372
commit 34d300d1da
176 changed files with 185 additions and 176 deletions
--- a/server/libs/infra/src/repositories/job.repository.ts
+++ b/server/libs/infra/src/repositories/job.repository.ts
@@ -0,0 +1,133 @@
+import {
+  IAssetJob,
+  IBaseJob,
+  IJobRepository,
+  IMetadataExtractionJob,
+  JobCounts,
+  JobItem,
+  JobName,
+  QueueName,
+} from '@app/domain';
+import { InjectQueue } from '@nestjs/bull';
+import { Logger } from '@nestjs/common';
+import { Queue, type JobCounts as BullJobCounts } from 'bull';
+
+export class JobRepository implements IJobRepository {
+  private logger = new Logger(JobRepository.name);
+  private queueMap: Record<QueueName, Queue> = {
+    [QueueName.STORAGE_TEMPLATE_MIGRATION]: this.storageTemplateMigration,
+    [QueueName.THUMBNAIL_GENERATION]: this.generateThumbnail,
+    [QueueName.METADATA_EXTRACTION]: this.metadataExtraction,
+    [QueueName.OBJECT_TAGGING]: this.objectTagging,
+    [QueueName.CLIP_ENCODING]: this.clipEmbedding,
+    [QueueName.VIDEO_CONVERSION]: this.videoTranscode,
+    [QueueName.BACKGROUND_TASK]: this.backgroundTask,
+    [QueueName.SEARCH]: this.searchIndex,
+  };
+
+  constructor(
+    @InjectQueue(QueueName.BACKGROUND_TASK) private backgroundTask: Queue,
+    @InjectQueue(QueueName.OBJECT_TAGGING) private objectTagging: Queue<IAssetJob | IBaseJob>,
+    @InjectQueue(QueueName.CLIP_ENCODING) private clipEmbedding: Queue<IAssetJob | IBaseJob>,
+    @InjectQueue(QueueName.METADATA_EXTRACTION) private metadataExtraction: Queue<IMetadataExtractionJob | IBaseJob>,
+    @InjectQueue(QueueName.STORAGE_TEMPLATE_MIGRATION) private storageTemplateMigration: Queue,
+    @InjectQueue(QueueName.THUMBNAIL_GENERATION) private generateThumbnail: Queue,
+    @InjectQueue(QueueName.VIDEO_CONVERSION) private videoTranscode: Queue<IAssetJob | IBaseJob>,
+    @InjectQueue(QueueName.SEARCH) private searchIndex: Queue,
+  ) {}
+
+  async isActive(name: QueueName): Promise<boolean> {
+    const counts = await this.getJobCounts(name);
+    return !!counts.active;
+  }
+
+  pause(name: QueueName) {
+    return this.queueMap[name].pause();
+  }
+
+  resume(name: QueueName) {
+    return this.queueMap[name].resume();
+  }
+
+  empty(name: QueueName) {
+    return this.queueMap[name].empty();
+  }
+
+  getJobCounts(name: QueueName): Promise<JobCounts> {
+    // Typecast needed because the `paused` key is missing from Bull's
+    // type definition. Can be removed once fixed upstream.
+    return this.queueMap[name].getJobCounts() as Promise<BullJobCounts & { paused: number }>;
+  }
+
+  async queue(item: JobItem): Promise<void> {
+    switch (item.name) {
+      case JobName.ASSET_UPLOADED:
+        await this.backgroundTask.add(item.name, item.data, { jobId: item.data.asset.id });
+        break;
+
+      case JobName.DELETE_FILES:
+        await this.backgroundTask.add(item.name, item.data);
+        break;
+
+      case JobName.QUEUE_OBJECT_TAGGING:
+      case JobName.DETECT_OBJECTS:
+      case JobName.CLASSIFY_IMAGE:
+        await this.objectTagging.add(item.name, item.data);
+        break;
+
+      case JobName.QUEUE_ENCODE_CLIP:
+      case JobName.ENCODE_CLIP:
+        await this.clipEmbedding.add(item.name, item.data);
+        break;
+
+      case JobName.QUEUE_METADATA_EXTRACTION:
+      case JobName.EXIF_EXTRACTION:
+      case JobName.EXTRACT_VIDEO_METADATA:
+      case JobName.REVERSE_GEOCODING:
+        await this.metadataExtraction.add(item.name, item.data);
+        break;
+
+      case JobName.QUEUE_GENERATE_THUMBNAILS:
+      case JobName.GENERATE_JPEG_THUMBNAIL:
+      case JobName.GENERATE_WEBP_THUMBNAIL:
+        await this.generateThumbnail.add(item.name, item.data);
+        break;
+
+      case JobName.USER_DELETION:
+        await this.backgroundTask.add(item.name, item.data);
+        break;
+
+      case JobName.STORAGE_TEMPLATE_MIGRATION:
+        await this.storageTemplateMigration.add(item.name);
+        break;
+
+      case JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE:
+        await this.storageTemplateMigration.add(item.name, item.data);
+        break;
+
+      case JobName.SYSTEM_CONFIG_CHANGE:
+        await this.backgroundTask.add(item.name, {});
+        break;
+
+      case JobName.QUEUE_VIDEO_CONVERSION:
+      case JobName.VIDEO_CONVERSION:
+        await this.videoTranscode.add(item.name, item.data);
+        break;
+
+      case JobName.SEARCH_INDEX_ASSETS:
+      case JobName.SEARCH_INDEX_ALBUMS:
+        await this.searchIndex.add(item.name, {});
+        break;
+
+      case JobName.SEARCH_INDEX_ASSET:
+      case JobName.SEARCH_INDEX_ALBUM:
+      case JobName.SEARCH_REMOVE_ALBUM:
+      case JobName.SEARCH_REMOVE_ASSET:
+        await this.searchIndex.add(item.name, item.data);
+        break;
+
+      default:
+        this.logger.error('Invalid job', item);
+    }
+  }
+}