Cutlery/immich
1
0
Fork 1
mirror of https://github.com/immich-app/immich.git synced 2025-11-04 03:27:09 -05:00
Code Issues Packages Projects Releases Wiki Activity

working addJobs

Browse Source
This commit is contained in:
mertalev 2025-04-30 21:54:44 -04:00
parent e773a7b7a1
commit 7bcf9aa3a7
No known key found for this signature in database
GPG Key ID: DF6ABC77AAD98C95
2 changed files with 19 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
server/src/repositories/job.repository.ts
View File

@ -1,9 +1,8 @@
import { Injectable } from '@nestjs/common'; import { Injectable } from '@nestjs/common';
import { ModuleRef, Reflector } from '@nestjs/core'; import { ModuleRef, Reflector } from '@nestjs/core';
import { ClassConstructor } from 'class-transformer'; import { ClassConstructor } from 'class-transformer';
import { AddJobsJobSpec, makeWorkerUtils, run, Runner, TaskSpec, WorkerUtils } from 'graphile-worker'; import { AddJobsJobSpec, makeWorkerUtils, run, Runner, WorkerUtils } from 'graphile-worker';
import { Kysely } from 'kysely'; import { Kysely } from 'kysely';
import { DateTime, Duration } from 'luxon';
import { InjectKysely } from 'nestjs-kysely'; import { InjectKysely } from 'nestjs-kysely';
import pg, { PoolConfig } from 'pg'; import pg, { PoolConfig } from 'pg';
import { DB } from 'src/db'; import { DB } from 'src/db';
@ -188,7 +187,7 @@ export class JobRepository {
} }
queue(item: JobItem): Promise<unknown> { queue(item: JobItem): Promise<unknown> {
return this.workerUtils!.addJob(this.getQueueName(item.name), item, this.getJobOptions(item)); return this.queueAll([item]);
} }
queueAll(items: JobItem[]): Promise<unknown> { queueAll(items: JobItem[]): Promise<unknown> {
@ -278,47 +277,27 @@ export class JobRepository {
} }
private getJobSpec(item: JobItem): AddJobsJobSpec { private getJobSpec(item: JobItem): AddJobsJobSpec {
const identifier = (this.handlers[item.name] as JobMapItem).queueName;
switch (item.name) { switch (item.name) {
case JobName.NOTIFY_ALBUM_UPDATE: { case JobName.NOTIFY_ALBUM_UPDATE: {
return { return {
identifier: item.name, identifier,
payload: item.data, payload: item,
jobKey: item.data.id, jobKey: item.data.id,
runAt: item.data?.delay ? new Date(Date.now() + item.data.delay) : undefined, runAt: item.data?.delay ? new Date(Date.now() + item.data.delay) : undefined,
}; };
} }
case JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE: { case JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE: {
return { identifier: item.name, payload: item.data, jobKey: QueueName.STORAGE_TEMPLATE_MIGRATION }; return { identifier, payload: item, jobKey: QueueName.STORAGE_TEMPLATE_MIGRATION };
} }
case JobName.GENERATE_PERSON_THUMBNAIL: { case JobName.GENERATE_PERSON_THUMBNAIL: {
return { identifier: item.name, payload: item.data, priority: 1 }; return { identifier, payload: item, priority: 1 };
} }
case JobName.QUEUE_FACIAL_RECOGNITION: { case JobName.QUEUE_FACIAL_RECOGNITION: {
return { identifier: item.name, payload: item.data, jobKey: JobName.QUEUE_FACIAL_RECOGNITION }; return { identifier, payload: item, jobKey: JobName.QUEUE_FACIAL_RECOGNITION };
} }
default: { default: {
return { identifier: item.name, payload: item.data }; return { identifier, payload: item };
}
}
}
private getJobOptions(item: JobItem): TaskSpec | undefined {
switch (item.name) {
case JobName.NOTIFY_ALBUM_UPDATE: {
let runAt: Date | undefined;
if (item.data?.delay) {
runAt = DateTime.now().plus(Duration.fromMillis(item.data.delay)).toJSDate();
}
return { jobKey: item.data.id, runAt };
}
case JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE: {
return { jobKey: QueueName.STORAGE_TEMPLATE_MIGRATION };
}
case JobName.GENERATE_PERSON_THUMBNAIL: {
return { priority: 1 };
}
case JobName.QUEUE_FACIAL_RECOGNITION: {
return { jobKey: JobName.QUEUE_FACIAL_RECOGNITION };
} }
} }
} }

17
server/src/services/metadata.service.ts
View File

@ -163,17 +163,20 @@ export class MetadataService extends BaseService {
async handleQueueMetadataExtraction(job: JobOf<JobName.QUEUE_METADATA_EXTRACTION>): Promise<JobStatus> { async handleQueueMetadataExtraction(job: JobOf<JobName.QUEUE_METADATA_EXTRACTION>): Promise<JobStatus> {
const { force } = job; const { force } = job;
let queue: { name: JobName.METADATA_EXTRACTION; data: { id: string } }[] = []; for (let i = 0; i < 10; i++) {
for await (const asset of this.assetJobRepository.streamForMetadataExtraction(force)) { let queue: { name: JobName.METADATA_EXTRACTION; data: { id: string } }[] = [];
queue.push({ name: JobName.METADATA_EXTRACTION, data: { id: asset.id } }); for await (const asset of this.assetJobRepository.streamForMetadataExtraction(force)) {
queue.push({ name: JobName.METADATA_EXTRACTION, data: { id: asset.id, source: 'upload' } as any });
if (queue.length >= JOBS_ASSET_PAGINATION_SIZE) { if (queue.length >= JOBS_ASSET_PAGINATION_SIZE) {
await this.jobRepository.queueAll(queue); await this.jobRepository.queueAll(queue);
queue = []; queue = [];
}
} }
await this.jobRepository.queueAll(queue);
} }
await this.jobRepository.queueAll(queue);
return JobStatus.SUCCESS; return JobStatus.SUCCESS;
} }