mirror of
https://github.com/immich-app/immich
synced 2025-11-07 17:27:20 +00:00
refactor: job names (#1343)
* refactor: job names * refactor: remove jobId
This commit is contained in:
parent
adacfb1110
commit
693adf8488
19 changed files with 125 additions and 289 deletions
|
|
@ -4,27 +4,22 @@ import {
|
|||
IMetadataExtractionJob,
|
||||
IThumbnailGenerationJob,
|
||||
IVideoTranscodeJob,
|
||||
assetUploadedProcessorName,
|
||||
exifExtractionProcessorName,
|
||||
generateJPEGThumbnailProcessorName,
|
||||
mp4ConversionProcessorName,
|
||||
videoMetadataExtractionProcessorName,
|
||||
QueueNameEnum,
|
||||
QueueName,
|
||||
JobName,
|
||||
} from '@app/job';
|
||||
import { InjectQueue, Process, Processor } from '@nestjs/bull';
|
||||
import { Job, Queue } from 'bull';
|
||||
import { randomUUID } from 'crypto';
|
||||
|
||||
@Processor(QueueNameEnum.ASSET_UPLOADED)
|
||||
@Processor(QueueName.ASSET_UPLOADED)
|
||||
export class AssetUploadedProcessor {
|
||||
constructor(
|
||||
@InjectQueue(QueueNameEnum.THUMBNAIL_GENERATION)
|
||||
@InjectQueue(QueueName.THUMBNAIL_GENERATION)
|
||||
private thumbnailGeneratorQueue: Queue<IThumbnailGenerationJob>,
|
||||
|
||||
@InjectQueue(QueueNameEnum.METADATA_EXTRACTION)
|
||||
@InjectQueue(QueueName.METADATA_EXTRACTION)
|
||||
private metadataExtractionQueue: Queue<IMetadataExtractionJob>,
|
||||
|
||||
@InjectQueue(QueueNameEnum.VIDEO_CONVERSION)
|
||||
@InjectQueue(QueueName.VIDEO_CONVERSION)
|
||||
private videoConversionQueue: Queue<IVideoTranscodeJob>,
|
||||
) {}
|
||||
|
||||
|
|
@ -37,30 +32,19 @@ export class AssetUploadedProcessor {
|
|||
*
|
||||
* @param job asset-uploaded
|
||||
*/
|
||||
@Process(assetUploadedProcessorName)
|
||||
@Process(JobName.ASSET_UPLOADED)
|
||||
async processUploadedVideo(job: Job<IAssetUploadedJob>) {
|
||||
const { asset, fileName } = job.data;
|
||||
|
||||
await this.thumbnailGeneratorQueue.add(generateJPEGThumbnailProcessorName, { asset }, { jobId: randomUUID() });
|
||||
await this.thumbnailGeneratorQueue.add(JobName.GENERATE_JPEG_THUMBNAIL, { asset });
|
||||
|
||||
// Video Conversion
|
||||
if (asset.type == AssetType.VIDEO) {
|
||||
await this.videoConversionQueue.add(mp4ConversionProcessorName, { asset }, { jobId: randomUUID() });
|
||||
await this.metadataExtractionQueue.add(
|
||||
videoMetadataExtractionProcessorName,
|
||||
{ asset, fileName },
|
||||
{ jobId: randomUUID() },
|
||||
);
|
||||
await this.videoConversionQueue.add(JobName.MP4_CONVERSION, { asset });
|
||||
await this.metadataExtractionQueue.add(JobName.EXTRACT_VIDEO_METADATA, { asset, fileName });
|
||||
} else {
|
||||
// Extract Metadata/Exif for Images - Currently the EXIF library on the web cannot extract EXIF for video yet
|
||||
await this.metadataExtractionQueue.add(
|
||||
exifExtractionProcessorName,
|
||||
{
|
||||
asset,
|
||||
fileName,
|
||||
},
|
||||
{ jobId: randomUUID() },
|
||||
);
|
||||
await this.metadataExtractionQueue.add(JobName.EXIF_EXTRACTION, { asset, fileName });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import { AssetEntity } from '@app/infra';
|
||||
import { QueueNameEnum } from '@app/job';
|
||||
import { QueueName } from '@app/job';
|
||||
import { Process, Processor } from '@nestjs/bull';
|
||||
import { Logger } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
|
|
@ -8,7 +8,7 @@ import fs from 'node:fs';
|
|||
import { FindOptionsWhere, IsNull, MoreThan, QueryFailedError, Repository } from 'typeorm';
|
||||
|
||||
// TODO: just temporary task to generate previous uploaded assets.
|
||||
@Processor(QueueNameEnum.CHECKSUM_GENERATION)
|
||||
@Processor(QueueName.CHECKSUM_GENERATION)
|
||||
export class GenerateChecksumProcessor {
|
||||
constructor(
|
||||
@InjectRepository(AssetEntity)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import { AssetEntity } from '@app/infra';
|
||||
import { SmartInfoEntity } from '@app/infra';
|
||||
import { MachineLearningJobNameEnum, QueueNameEnum } from '@app/job';
|
||||
import { QueueName, JobName } from '@app/job';
|
||||
import { IMachineLearningJob } from '@app/job/interfaces/machine-learning.interface';
|
||||
import { Process, Processor } from '@nestjs/bull';
|
||||
import { Logger } from '@nestjs/common';
|
||||
|
|
@ -11,14 +11,14 @@ import { Repository } from 'typeorm';
|
|||
|
||||
const immich_machine_learning_url = process.env.IMMICH_MACHINE_LEARNING_URL || 'http://immich-machine-learning:3003';
|
||||
|
||||
@Processor(QueueNameEnum.MACHINE_LEARNING)
|
||||
@Processor(QueueName.MACHINE_LEARNING)
|
||||
export class MachineLearningProcessor {
|
||||
constructor(
|
||||
@InjectRepository(SmartInfoEntity)
|
||||
private smartInfoRepository: Repository<SmartInfoEntity>,
|
||||
) {}
|
||||
|
||||
@Process({ name: MachineLearningJobNameEnum.IMAGE_TAGGING, concurrency: 2 })
|
||||
@Process({ name: JobName.IMAGE_TAGGING, concurrency: 2 })
|
||||
async tagImage(job: Job<IMachineLearningJob>) {
|
||||
const { asset } = job.data;
|
||||
|
||||
|
|
@ -37,7 +37,7 @@ export class MachineLearningProcessor {
|
|||
}
|
||||
}
|
||||
|
||||
@Process({ name: MachineLearningJobNameEnum.OBJECT_DETECTION, concurrency: 2 })
|
||||
@Process({ name: JobName.OBJECT_DETECTION, concurrency: 2 })
|
||||
async detectObject(job: Job<IMachineLearningJob>) {
|
||||
try {
|
||||
const { asset }: { asset: AssetEntity } = job.data;
|
||||
|
|
|
|||
|
|
@ -2,11 +2,9 @@ import { AssetEntity, ExifEntity } from '@app/infra';
|
|||
import {
|
||||
IExifExtractionProcessor,
|
||||
IVideoLengthExtractionProcessor,
|
||||
exifExtractionProcessorName,
|
||||
videoMetadataExtractionProcessorName,
|
||||
reverseGeocodingProcessorName,
|
||||
IReverseGeocodingProcessor,
|
||||
QueueNameEnum,
|
||||
QueueName,
|
||||
JobName,
|
||||
} from '@app/job';
|
||||
import { Process, Processor } from '@nestjs/bull';
|
||||
import { Logger } from '@nestjs/common';
|
||||
|
|
@ -73,7 +71,7 @@ export type GeoData = {
|
|||
distance: number;
|
||||
};
|
||||
|
||||
@Processor(QueueNameEnum.METADATA_EXTRACTION)
|
||||
@Processor(QueueName.METADATA_EXTRACTION)
|
||||
export class MetadataExtractionProcessor {
|
||||
private logger = new Logger(MetadataExtractionProcessor.name);
|
||||
private isGeocodeInitialized = false;
|
||||
|
|
@ -140,7 +138,7 @@ export class MetadataExtractionProcessor {
|
|||
return { country, state, city };
|
||||
}
|
||||
|
||||
@Process(exifExtractionProcessorName)
|
||||
@Process(JobName.EXIF_EXTRACTION)
|
||||
async extractExifInfo(job: Job<IExifExtractionProcessor>) {
|
||||
try {
|
||||
const { asset, fileName }: { asset: AssetEntity; fileName: string } = job.data;
|
||||
|
|
@ -262,7 +260,7 @@ export class MetadataExtractionProcessor {
|
|||
}
|
||||
}
|
||||
|
||||
@Process({ name: reverseGeocodingProcessorName })
|
||||
@Process({ name: JobName.REVERSE_GEOCODING })
|
||||
async reverseGeocoding(job: Job<IReverseGeocodingProcessor>) {
|
||||
if (this.isGeocodeInitialized) {
|
||||
const { latitude, longitude } = job.data;
|
||||
|
|
@ -271,7 +269,7 @@ export class MetadataExtractionProcessor {
|
|||
}
|
||||
}
|
||||
|
||||
@Process({ name: videoMetadataExtractionProcessorName, concurrency: 2 })
|
||||
@Process({ name: JobName.EXTRACT_VIDEO_METADATA, concurrency: 2 })
|
||||
async extractVideoMetadata(job: Job<IVideoLengthExtractionProcessor>) {
|
||||
const { asset, fileName } = job.data;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,14 +1,14 @@
|
|||
import { APP_UPLOAD_LOCATION } from '@app/common';
|
||||
import { AssetEntity } from '@app/infra';
|
||||
import { ImmichConfigService } from '@app/immich-config';
|
||||
import { QueueNameEnum, templateMigrationProcessorName, updateTemplateProcessorName } from '@app/job';
|
||||
import { QueueName, JobName } from '@app/job';
|
||||
import { StorageService } from '@app/storage';
|
||||
import { Process, Processor } from '@nestjs/bull';
|
||||
import { Logger } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Repository } from 'typeorm';
|
||||
|
||||
@Processor(QueueNameEnum.STORAGE_MIGRATION)
|
||||
@Processor(QueueName.CONFIG)
|
||||
export class StorageMigrationProcessor {
|
||||
readonly logger: Logger = new Logger(StorageMigrationProcessor.name);
|
||||
|
||||
|
|
@ -24,7 +24,7 @@ export class StorageMigrationProcessor {
|
|||
* Migration process when a new user set a new storage template.
|
||||
* @param job
|
||||
*/
|
||||
@Process({ name: templateMigrationProcessorName, concurrency: 100 })
|
||||
@Process({ name: JobName.TEMPLATE_MIGRATION, concurrency: 100 })
|
||||
async templateMigration() {
|
||||
console.time('migrating-time');
|
||||
const assets = await this.assetRepository.find({
|
||||
|
|
@ -54,7 +54,7 @@ export class StorageMigrationProcessor {
|
|||
* This is to ensure the synchronization between processes.
|
||||
* @param job
|
||||
*/
|
||||
@Process({ name: updateTemplateProcessorName, concurrency: 1 })
|
||||
@Process({ name: JobName.CONFIG_CHANGE, concurrency: 1 })
|
||||
async updateTemplate() {
|
||||
await this.immichConfigService.refreshConfig();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,20 +1,12 @@
|
|||
import { APP_UPLOAD_LOCATION } from '@app/common';
|
||||
import { AssetEntity, AssetType } from '@app/infra';
|
||||
import {
|
||||
WebpGeneratorProcessor,
|
||||
generateJPEGThumbnailProcessorName,
|
||||
generateWEBPThumbnailProcessorName,
|
||||
JpegGeneratorProcessor,
|
||||
QueueNameEnum,
|
||||
MachineLearningJobNameEnum,
|
||||
} from '@app/job';
|
||||
import { WebpGeneratorProcessor, JpegGeneratorProcessor, QueueName, JobName } from '@app/job';
|
||||
import { InjectQueue, Process, Processor } from '@nestjs/bull';
|
||||
import { Logger } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { mapAsset } from 'apps/immich/src/api-v1/asset/response-dto/asset-response.dto';
|
||||
import { Job, Queue } from 'bull';
|
||||
import ffmpeg from 'fluent-ffmpeg';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { existsSync, mkdirSync } from 'node:fs';
|
||||
import sanitize from 'sanitize-filename';
|
||||
import sharp from 'sharp';
|
||||
|
|
@ -23,7 +15,7 @@ import { join } from 'path';
|
|||
import { CommunicationGateway } from 'apps/immich/src/api-v1/communication/communication.gateway';
|
||||
import { IMachineLearningJob } from '@app/job/interfaces/machine-learning.interface';
|
||||
|
||||
@Processor(QueueNameEnum.THUMBNAIL_GENERATION)
|
||||
@Processor(QueueName.THUMBNAIL_GENERATION)
|
||||
export class ThumbnailGeneratorProcessor {
|
||||
readonly logger: Logger = new Logger(ThumbnailGeneratorProcessor.name);
|
||||
|
||||
|
|
@ -31,16 +23,16 @@ export class ThumbnailGeneratorProcessor {
|
|||
@InjectRepository(AssetEntity)
|
||||
private assetRepository: Repository<AssetEntity>,
|
||||
|
||||
@InjectQueue(QueueNameEnum.THUMBNAIL_GENERATION)
|
||||
@InjectQueue(QueueName.THUMBNAIL_GENERATION)
|
||||
private thumbnailGeneratorQueue: Queue,
|
||||
|
||||
private wsCommunicationGateway: CommunicationGateway,
|
||||
|
||||
@InjectQueue(QueueNameEnum.MACHINE_LEARNING)
|
||||
@InjectQueue(QueueName.MACHINE_LEARNING)
|
||||
private machineLearningQueue: Queue<IMachineLearningJob>,
|
||||
) {}
|
||||
|
||||
@Process({ name: generateJPEGThumbnailProcessorName, concurrency: 3 })
|
||||
@Process({ name: JobName.GENERATE_JPEG_THUMBNAIL, concurrency: 3 })
|
||||
async generateJPEGThumbnail(job: Job<JpegGeneratorProcessor>) {
|
||||
const basePath = APP_UPLOAD_LOCATION;
|
||||
|
||||
|
|
@ -70,13 +62,10 @@ export class ThumbnailGeneratorProcessor {
|
|||
// Update resize path to send to generate webp queue
|
||||
asset.resizePath = jpegThumbnailPath;
|
||||
|
||||
await this.thumbnailGeneratorQueue.add(generateWEBPThumbnailProcessorName, { asset }, { jobId: randomUUID() });
|
||||
await this.machineLearningQueue.add(MachineLearningJobNameEnum.IMAGE_TAGGING, { asset }, { jobId: randomUUID() });
|
||||
await this.machineLearningQueue.add(
|
||||
MachineLearningJobNameEnum.OBJECT_DETECTION,
|
||||
{ asset },
|
||||
{ jobId: randomUUID() },
|
||||
);
|
||||
await this.thumbnailGeneratorQueue.add(JobName.GENERATE_WEBP_THUMBNAIL, { asset });
|
||||
await this.machineLearningQueue.add(JobName.IMAGE_TAGGING, { asset });
|
||||
await this.machineLearningQueue.add(JobName.OBJECT_DETECTION, { asset });
|
||||
|
||||
this.wsCommunicationGateway.server.to(asset.userId).emit('on_upload_success', JSON.stringify(mapAsset(asset)));
|
||||
}
|
||||
|
||||
|
|
@ -104,19 +93,15 @@ export class ThumbnailGeneratorProcessor {
|
|||
// Update resize path to send to generate webp queue
|
||||
asset.resizePath = jpegThumbnailPath;
|
||||
|
||||
await this.thumbnailGeneratorQueue.add(generateWEBPThumbnailProcessorName, { asset }, { jobId: randomUUID() });
|
||||
await this.machineLearningQueue.add(MachineLearningJobNameEnum.IMAGE_TAGGING, { asset }, { jobId: randomUUID() });
|
||||
await this.machineLearningQueue.add(
|
||||
MachineLearningJobNameEnum.OBJECT_DETECTION,
|
||||
{ asset },
|
||||
{ jobId: randomUUID() },
|
||||
);
|
||||
await this.thumbnailGeneratorQueue.add(JobName.GENERATE_WEBP_THUMBNAIL, { asset });
|
||||
await this.machineLearningQueue.add(JobName.IMAGE_TAGGING, { asset });
|
||||
await this.machineLearningQueue.add(JobName.OBJECT_DETECTION, { asset });
|
||||
|
||||
this.wsCommunicationGateway.server.to(asset.userId).emit('on_upload_success', JSON.stringify(mapAsset(asset)));
|
||||
}
|
||||
}
|
||||
|
||||
@Process({ name: generateWEBPThumbnailProcessorName, concurrency: 3 })
|
||||
@Process({ name: JobName.GENERATE_WEBP_THUMBNAIL, concurrency: 3 })
|
||||
async generateWepbThumbnail(job: Job<WebpGeneratorProcessor>) {
|
||||
const { asset } = job.data;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import { APP_UPLOAD_LOCATION, userUtils } from '@app/common';
|
||||
import { APIKeyEntity, AssetEntity, UserEntity } from '@app/infra';
|
||||
import { QueueNameEnum, userDeletionProcessorName } from '@app/job';
|
||||
import { QueueName, JobName } from '@app/job';
|
||||
import { IUserDeletionJob } from '@app/job/interfaces/user-deletion.interface';
|
||||
import { Process, Processor } from '@nestjs/bull';
|
||||
import { Logger } from '@nestjs/common';
|
||||
|
|
@ -10,7 +10,7 @@ import { join } from 'path';
|
|||
import fs from 'fs';
|
||||
import { Repository } from 'typeorm';
|
||||
|
||||
@Processor(QueueNameEnum.USER_DELETION)
|
||||
@Processor(QueueName.USER_DELETION)
|
||||
export class UserDeletionProcessor {
|
||||
private logger = new Logger(UserDeletionProcessor.name);
|
||||
|
||||
|
|
@ -25,7 +25,7 @@ export class UserDeletionProcessor {
|
|||
private apiKeyRepository: Repository<APIKeyEntity>,
|
||||
) {}
|
||||
|
||||
@Process(userDeletionProcessorName)
|
||||
@Process(JobName.USER_DELETION)
|
||||
async processUserDeletion(job: Job<IUserDeletionJob>) {
|
||||
const { user } = job.data;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
import { APP_UPLOAD_LOCATION } from '@app/common/constants';
|
||||
import { AssetEntity } from '@app/infra';
|
||||
import { QueueNameEnum } from '@app/job';
|
||||
import { mp4ConversionProcessorName } from '@app/job/constants/job-name.constant';
|
||||
import { QueueName, JobName } from '@app/job';
|
||||
import { IMp4ConversionProcessor } from '@app/job/interfaces/video-transcode.interface';
|
||||
import { Process, Processor } from '@nestjs/bull';
|
||||
import { Logger } from '@nestjs/common';
|
||||
|
|
@ -12,7 +11,7 @@ import { existsSync, mkdirSync } from 'fs';
|
|||
import { ImmichConfigService } from 'libs/immich-config/src';
|
||||
import { Repository } from 'typeorm';
|
||||
|
||||
@Processor(QueueNameEnum.VIDEO_CONVERSION)
|
||||
@Processor(QueueName.VIDEO_CONVERSION)
|
||||
export class VideoTranscodeProcessor {
|
||||
constructor(
|
||||
@InjectRepository(AssetEntity)
|
||||
|
|
@ -20,7 +19,7 @@ export class VideoTranscodeProcessor {
|
|||
private immichConfigService: ImmichConfigService,
|
||||
) {}
|
||||
|
||||
@Process({ name: mp4ConversionProcessorName, concurrency: 2 })
|
||||
@Process({ name: JobName.MP4_CONVERSION, concurrency: 2 })
|
||||
async mp4Conversion(job: Job<IMp4ConversionProcessor>) {
|
||||
const { asset } = job.data;
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue