c-l/immich
1
0
Fork 0
mirror of https://github.com/immich-app/immich.git synced 2025-01-01 08:31:59 +00:00
Code Releases Activity

feat: run all offline checks in a single job

Browse source
This commit is contained in:
Jonathan Jogenfors 2024-12-02 22:35:51 +01:00
parent ba71fd42da
commit 8ecde3b277
3 changed files with 61 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
server/src/interfaces/job.interface.ts
View file

@ -85,7 +85,7 @@ export enum JobName {
LIBRARY_QUEUE_SYNC_FILES = 'library-queue-sync-files',
LIBRARY_QUEUE_SYNC_ASSETS = 'library-queue-sync-assets',
LIBRARY_SYNC_FILE = 'library-sync-file',
LIBRARY_SYNC_ASSET = 'library-sync-asset',
LIBRARY_SYNC_ASSETS = 'library-sync-assets',
LIBRARY_DELETE = 'library-delete',
LIBRARY_QUEUE_SYNC_ALL = 'library-queue-sync-all',
LIBRARY_QUEUE_CLEANUP = 'library-queue-cleanup',
@ -148,15 +148,15 @@ export interface ILibraryFileJob extends IEntityJob {
assetPath: string;
}
export interface ILibraryAssetJob extends IEntityJob {
importPaths: string[];
exclusionPatterns: string[];
}
export interface IBulkEntityJob extends IBaseJob {
ids: string[];
}
export interface ILibraryAssetsJob extends IBulkEntityJob {
importPaths: string[];
exclusionPatterns: string[];
}
export interface IDeleteFilesJob extends IBaseJob {
files: Array<string | null | undefined>;
}
@ -287,7 +287,7 @@ export type JobItem =
| { name: JobName.LIBRARY_SYNC_FILE; data: ILibraryFileJob }
| { name: JobName.LIBRARY_QUEUE_SYNC_FILES; data: IEntityJob }
| { name: JobName.LIBRARY_QUEUE_SYNC_ASSETS; data: IEntityJob }
| { name: JobName.LIBRARY_SYNC_ASSET; data: ILibraryAssetJob }
| { name: JobName.LIBRARY_SYNC_ASSETS; data: ILibraryAssetsJob }
| { name: JobName.LIBRARY_DELETE; data: IEntityJob }
| { name: JobName.LIBRARY_QUEUE_SYNC_ALL; data?: IBaseJob }
| { name: JobName.LIBRARY_QUEUE_CLEANUP; data: IBaseJob }

56
server/src/services/library.service.spec.ts
View file

@ -10,7 +10,7 @@ import { ICronRepository } from 'src/interfaces/cron.interface';
import { IDatabaseRepository } from 'src/interfaces/database.interface';
import {
IJobRepository,
ILibraryAssetJob,
ILibraryAssetsJob,
ILibraryFileJob,
JobName,
JOBS_LIBRARY_PAGINATION_SIZE,
@ -231,7 +231,7 @@ describe(LibraryService.name, () => {
expect(jobMock.queueAll).toHaveBeenCalledWith([
{
name: JobName.LIBRARY_SYNC_ASSET,
name: JobName.LIBRARY_SYNC_ASSETS,
data: {
id: assetStub.external.id,
importPaths: libraryStub.externalLibrary1.importPaths,
@ -250,22 +250,22 @@ describe(LibraryService.name, () => {
describe('handleSyncAsset', () => {
it('should skip missing assets', async () => {
const mockAssetJob: ILibraryAssetJob = {
id: assetStub.external.id,
const mockAssetJob: ILibraryAssetsJob = {
ids: [assetStub.external.id],
importPaths: ['/'],
exclusionPatterns: [],
};
assetMock.getById.mockResolvedValue(null);
await expect(sut.handleSyncAsset(mockAssetJob)).resolves.toBe(JobStatus.SKIPPED);
await expect(sut.handleSyncAssets(mockAssetJob)).resolves.toBe(JobStatus.SKIPPED);
expect(assetMock.remove).not.toHaveBeenCalled();
});
it('should offline assets no longer on disk', async () => {
const mockAssetJob: ILibraryAssetJob = {
id: assetStub.external.id,
const mockAssetJob: ILibraryAssetsJob = {
ids: [assetStub.external.id],
importPaths: ['/'],
exclusionPatterns: [],
};
@ -273,7 +273,7 @@ describe(LibraryService.name, () => {
assetMock.getById.mockResolvedValue(assetStub.external);
storageMock.stat.mockRejectedValue(new Error('ENOENT, no such file or directory'));
await expect(sut.handleSyncAsset(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
await expect(sut.handleSyncAssets(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
expect(assetMock.updateAll).toHaveBeenCalledWith([assetStub.external.id], {
isOffline: true,
@ -282,15 +282,15 @@ describe(LibraryService.name, () => {
});
it('should offline assets matching an exclusion pattern', async () => {
const mockAssetJob: ILibraryAssetJob = {
id: assetStub.external.id,
const mockAssetJob: ILibraryAssetsJob = {
ids: [assetStub.external.id],
importPaths: ['/'],
exclusionPatterns: ['**/user1/**'],
};
assetMock.getById.mockResolvedValue(assetStub.external);
await expect(sut.handleSyncAsset(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
await expect(sut.handleSyncAssets(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
expect(assetMock.updateAll).toHaveBeenCalledWith([assetStub.external.id], {
isOffline: true,
deletedAt: expect.any(Date),
@ -298,8 +298,8 @@ describe(LibraryService.name, () => {
});
it('should set assets outside of import paths as offline', async () => {
const mockAssetJob: ILibraryAssetJob = {
id: assetStub.external.id,
const mockAssetJob: ILibraryAssetsJob = {
ids: [assetStub.external.id],
importPaths: ['/data/user2'],
exclusionPatterns: [],
};
@ -307,7 +307,7 @@ describe(LibraryService.name, () => {
assetMock.getById.mockResolvedValue(assetStub.external);
storageMock.checkFileExists.mockResolvedValue(true);
await expect(sut.handleSyncAsset(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
await expect(sut.handleSyncAssets(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
expect(assetMock.updateAll).toHaveBeenCalledWith([assetStub.external.id], {
isOffline: true,
@ -316,8 +316,8 @@ describe(LibraryService.name, () => {
});
it('should do nothing with online assets', async () => {
const mockAssetJob: ILibraryAssetJob = {
id: assetStub.external.id,
const mockAssetJob: ILibraryAssetsJob = {
ids: [assetStub.external.id],
importPaths: ['/'],
exclusionPatterns: [],
};
@ -325,14 +325,14 @@ describe(LibraryService.name, () => {
assetMock.getById.mockResolvedValue(assetStub.external);
storageMock.stat.mockResolvedValue({ mtime: assetStub.external.fileModifiedAt } as Stats);
await expect(sut.handleSyncAsset(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
await expect(sut.handleSyncAssets(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
expect(assetMock.updateAll).not.toHaveBeenCalled();
});
it('should un-trash an asset previously marked as offline', async () => {
const mockAssetJob: ILibraryAssetJob = {
id: assetStub.external.id,
const mockAssetJob: ILibraryAssetsJob = {
ids: [assetStub.external.id],
importPaths: ['/'],
exclusionPatterns: [],
};
@ -340,7 +340,7 @@ describe(LibraryService.name, () => {
assetMock.getById.mockResolvedValue(assetStub.trashedOffline);
storageMock.stat.mockResolvedValue({ mtime: assetStub.trashedOffline.fileModifiedAt } as Stats);
await expect(sut.handleSyncAsset(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
await expect(sut.handleSyncAssets(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
expect(assetMock.updateAll).toHaveBeenCalledWith([assetStub.trashedOffline.id], {
deletedAt: null,
@ -353,8 +353,8 @@ describe(LibraryService.name, () => {
});
it('should update file when mtime has changed', async () => {
const mockAssetJob: ILibraryAssetJob = {
id: assetStub.external.id,
const mockAssetJob: ILibraryAssetsJob = {
ids: [assetStub.external.id],
importPaths: ['/'],
exclusionPatterns: [],
};
@ -363,7 +363,7 @@ describe(LibraryService.name, () => {
assetMock.getById.mockResolvedValue(assetStub.external);
storageMock.stat.mockResolvedValue({ mtime: newMTime } as Stats);
await expect(sut.handleSyncAsset(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
await expect(sut.handleSyncAssets(mockAssetJob)).resolves.toBe(JobStatus.SUCCESS);
expect(assetMock.updateAll).toHaveBeenCalledWith([assetStub.external.id], {
fileModifiedAt: newMTime,
@ -969,7 +969,7 @@ describe(LibraryService.name, () => {
},
]);
expect(jobMock.queueAll).toHaveBeenCalledWith([
{ name: JobName.LIBRARY_SYNC_ASSET, data: expect.objectContaining({ id: assetStub.image.id }) },
{ name: JobName.LIBRARY_SYNC_ASSETS, data: expect.objectContaining({ ids: [assetStub.image.id] }) },
]);
});
@ -994,7 +994,7 @@ describe(LibraryService.name, () => {
},
]);
expect(jobMock.queueAll).toHaveBeenCalledWith([
{ name: JobName.LIBRARY_SYNC_ASSET, data: expect.objectContaining({ id: assetStub.image.id }) },
{ name: JobName.LIBRARY_SYNC_ASSETS, data: expect.objectContaining({ id: [assetStub.image.id] }) },
]);
});
@ -1009,7 +1009,7 @@ describe(LibraryService.name, () => {
await sut.watchAll();
expect(jobMock.queueAll).toHaveBeenCalledWith([
{ name: JobName.LIBRARY_SYNC_ASSET, data: expect.objectContaining({ id: assetStub.image.id }) },
{ name: JobName.LIBRARY_SYNC_ASSETS, data: expect.objectContaining({ ids: [assetStub.image.id] }) },
]);
});
@ -1166,9 +1166,9 @@ describe(LibraryService.name, () => {
expect(jobMock.queueAll).toHaveBeenCalledWith([
{
name: JobName.LIBRARY_SYNC_ASSET,
name: JobName.LIBRARY_SYNC_ASSETS,
data: {
id: assetStub.image1.id,
ids: [assetStub.image1.id],
importPaths: libraryStub.externalLibrary1.importPaths,
exclusionPatterns: libraryStub.externalLibrary1.exclusionPatterns,
},

44
server/src/services/library.service.ts
View file

@ -242,12 +242,10 @@ export class LibraryService extends BaseService {
}
private async syncAssets({ importPaths, exclusionPatterns }: LibraryEntity, assetIds: string[]) {
await this.jobRepository.queueAll(
assetIds.map((assetId) => ({
name: JobName.LIBRARY_SYNC_ASSET,
data: { id: assetId, importPaths, exclusionPatterns },
})),
);
await this.jobRepository.queue({
name: JobName.LIBRARY_SYNC_ASSETS,
data: { ids: assetIds, importPaths, exclusionPatterns },
});
}
private async validateImportPath(importPath: string): Promise<ValidateLibraryImportPathResponseDto> {
@ -472,27 +470,35 @@ export class LibraryService extends BaseService {
return JobStatus.SUCCESS;
}
@OnJob({ name: JobName.LIBRARY_SYNC_ASSET, queue: QueueName.LIBRARY })
async handleSyncAsset(job: JobOf<JobName.LIBRARY_SYNC_ASSET>): Promise<JobStatus> {
const asset = await this.assetRepository.getById(job.id);
@OnJob({ name: JobName.LIBRARY_SYNC_ASSETS, queue: QueueName.LIBRARY })
async handleSyncAssets(job: JobOf<JobName.LIBRARY_SYNC_ASSETS>): Promise<JobStatus> {
for (const id of job.ids) {
await this.handleSyncAsset(id, job.importPaths, job.exclusionPatterns);
}
return JobStatus.SUCCESS;
}
private async handleSyncAsset(id: string, importPaths: string[], exclusionPatterns: string[]): Promise<JobStatus> {
const asset = await this.assetRepository.getById(id);
if (!asset) {
return JobStatus.SKIPPED;
}
const markOffline = async (explanation: string) => {
if (!asset.isOffline) {
this.logger.debug(`${explanation}, removing: ${asset.originalPath}`);
this.logger.debug(`${explanation}, moving to trash: ${asset.originalPath}`);
await this.assetRepository.updateAll([asset.id], { isOffline: true, deletedAt: new Date() });
}
};
const isInPath = job.importPaths.find((path) => asset.originalPath.startsWith(path));
const isInPath = importPaths.find((path) => asset.originalPath.startsWith(path));
if (!isInPath) {
await markOffline('Asset is no longer in an import path');
return JobStatus.SUCCESS;
}
const isExcluded = job.exclusionPatterns.some((pattern) => picomatch.isMatch(asset.originalPath, pattern));
const isExcluded = exclusionPatterns.some((pattern) => picomatch.isMatch(asset.originalPath, pattern));
if (isExcluded) {
await markOffline('Asset is covered by an exclusion pattern');
return JobStatus.SUCCESS;
@ -597,12 +603,14 @@ export class LibraryService extends BaseService {
for await (const assets of onlineAssets) {
assetCount += assets.length;
this.logger.debug(`Discovered ${assetCount} asset(s) in library ${library.id}...`);
await this.jobRepository.queueAll(
assets.map((asset) => ({
name: JobName.LIBRARY_SYNC_ASSET,
data: { id: asset.id, importPaths: library.importPaths, exclusionPatterns: library.exclusionPatterns },
})),
);
await this.jobRepository.queue({
name: JobName.LIBRARY_SYNC_ASSETS,
data: {
ids: assets.map((asset) => asset.id),
importPaths: library.importPaths,
exclusionPatterns: library.exclusionPatterns,
},
});
this.logger.debug(`Queued check of ${assets.length} asset(s) in library ${library.id}...`);
}