mirror of
https://github.com/immich-app/immich
synced 2026-08-15 13:03:57 +00:00
feat: integrity check jobs (missing files, untracked files, checksums) (#24205)
Co-authored-by: Daniel Dietzler <mail@ddietzler.dev> Signed-off-by: izzy <me@insrt.uk>
This commit is contained in:
Paul Makles
GitHub
parent
4ead3e697d
commit
74878628c8
66 changed files with 5711 additions and 27 deletions
|
|
@ -50,6 +50,22 @@ export type SystemConfig = {
|
|||
enabled: boolean;
|
||||
};
|
||||
};
|
||||
integrityChecks: {
|
||||
missingFiles: {
|
||||
enabled: boolean;
|
||||
cronExpression: string;
|
||||
};
|
||||
untrackedFiles: {
|
||||
enabled: boolean;
|
||||
cronExpression: string;
|
||||
};
|
||||
checksumFiles: {
|
||||
enabled: boolean;
|
||||
cronExpression: string;
|
||||
timeLimit: number;
|
||||
percentageLimit: number;
|
||||
};
|
||||
};
|
||||
job: Record<ConcurrentQueueName, { concurrency: number }>;
|
||||
logging: {
|
||||
enabled: boolean;
|
||||
|
|
@ -233,6 +249,22 @@ export const defaults = Object.freeze<SystemConfig>({
|
|||
enabled: false,
|
||||
},
|
||||
},
|
||||
integrityChecks: {
|
||||
missingFiles: {
|
||||
enabled: true,
|
||||
cronExpression: CronExpression.EVERY_DAY_AT_3AM,
|
||||
},
|
||||
untrackedFiles: {
|
||||
enabled: true,
|
||||
cronExpression: CronExpression.EVERY_DAY_AT_3AM,
|
||||
},
|
||||
checksumFiles: {
|
||||
enabled: true,
|
||||
cronExpression: CronExpression.EVERY_DAY_AT_3AM,
|
||||
timeLimit: 60 * 60 * 1000, // 1 hour
|
||||
percentageLimit: 1, // 100% of assets
|
||||
},
|
||||
},
|
||||
job: {
|
||||
[QueueName.BackgroundTask]: { concurrency: 5 },
|
||||
[QueueName.SmartSearch]: { concurrency: 2 },
|
||||
|
|
@ -247,6 +279,7 @@ export const defaults = Object.freeze<SystemConfig>({
|
|||
[QueueName.Notification]: { concurrency: 5 },
|
||||
[QueueName.Ocr]: { concurrency: 1 },
|
||||
[QueueName.Workflow]: { concurrency: 5 },
|
||||
[QueueName.IntegrityCheck]: { concurrency: 1 },
|
||||
[QueueName.Editor]: { concurrency: 2 },
|
||||
},
|
||||
logging: {
|
||||
|
|
|
|||
|
|
@ -156,6 +156,7 @@ export const endpointTags: Record<ApiTag, string> = {
|
|||
[ApiTag.Duplicates]: 'Endpoints for managing and identifying duplicate assets.',
|
||||
[ApiTag.Faces]:
|
||||
'A face is a detected human face within an asset, which can be associated with a person. Faces are normally detected via machine learning, but can also be created via manually.',
|
||||
[ApiTag.Integrity]: 'Endpoints for viewing and managing integrity reports.',
|
||||
[ApiTag.Jobs]:
|
||||
'Queues and background jobs are used for processing tasks asynchronously. Queues can be paused and resumed as needed.',
|
||||
[ApiTag.Libraries]:
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import { DatabaseBackupController } from 'src/controllers/database-backup.contro
|
|||
import { DownloadController } from 'src/controllers/download.controller';
|
||||
import { DuplicateController } from 'src/controllers/duplicate.controller';
|
||||
import { FaceController } from 'src/controllers/face.controller';
|
||||
import { IntegrityAdminController } from 'src/controllers/integrity-admin.controller';
|
||||
import { JobController } from 'src/controllers/job.controller';
|
||||
import { LibraryController } from 'src/controllers/library.controller';
|
||||
import { MaintenanceController } from 'src/controllers/maintenance.controller';
|
||||
|
|
@ -52,6 +53,7 @@ export const controllers = [
|
|||
DownloadController,
|
||||
DuplicateController,
|
||||
FaceController,
|
||||
IntegrityAdminController,
|
||||
JobController,
|
||||
LibraryController,
|
||||
MaintenanceController,
|
||||
|
|
|
|||
91
server/src/controllers/integrity-admin.controller.ts
Normal file
91
server/src/controllers/integrity-admin.controller.ts
Normal file
|
|
@ -0,0 +1,91 @@
|
|||
import { Controller, Delete, Get, HttpCode, HttpStatus, Next, Param, Query, Res } from '@nestjs/common';
|
||||
import { ApiTags } from '@nestjs/swagger';
|
||||
import { NextFunction, Response } from 'express';
|
||||
import { Endpoint, HistoryBuilder } from 'src/decorators';
|
||||
import { AuthDto } from 'src/dtos/auth.dto';
|
||||
import {
|
||||
IntegrityGetReportDto,
|
||||
IntegrityReportResponseDto,
|
||||
IntegrityReportSummaryResponseDto,
|
||||
} from 'src/dtos/integrity.dto';
|
||||
import { ApiTag, Permission } from 'src/enum';
|
||||
import { Auth, Authenticated, FileResponse } from 'src/middleware/auth.guard';
|
||||
import { LoggingRepository } from 'src/repositories/logging.repository';
|
||||
import { IntegrityService } from 'src/services/integrity.service';
|
||||
import { sendFile } from 'src/utils/file';
|
||||
import { IntegrityReportTypeParamDto, UUIDv7ParamDto } from 'src/validation';
|
||||
|
||||
@ApiTags(ApiTag.Maintenance)
|
||||
@Controller('admin/integrity')
|
||||
export class IntegrityAdminController {
|
||||
constructor(
|
||||
private logger: LoggingRepository,
|
||||
private service: IntegrityService,
|
||||
) {}
|
||||
|
||||
@Get('summary')
|
||||
@Endpoint({
|
||||
summary: 'Get integrity report summary',
|
||||
description: 'Get a count of the items flagged in each integrity report',
|
||||
history: new HistoryBuilder().added('v3.0.0').alpha('v3.0.0'),
|
||||
})
|
||||
@Authenticated({ permission: Permission.Maintenance, admin: true })
|
||||
getIntegrityReportSummary(): Promise<IntegrityReportSummaryResponseDto> {
|
||||
return this.service.getIntegrityReportSummary();
|
||||
}
|
||||
|
||||
@Get('report')
|
||||
@HttpCode(HttpStatus.OK)
|
||||
@Endpoint({
|
||||
summary: 'Get integrity report by type',
|
||||
description: 'Get all flagged items by integrity report type',
|
||||
history: new HistoryBuilder().added('v3.0.0').alpha('v3.0.0'),
|
||||
})
|
||||
@Authenticated({ permission: Permission.Maintenance, admin: true })
|
||||
getIntegrityReport(@Query() dto: IntegrityGetReportDto): Promise<IntegrityReportResponseDto> {
|
||||
return this.service.getIntegrityReport(dto);
|
||||
}
|
||||
|
||||
@Get('report/:id/file')
|
||||
@Endpoint({
|
||||
summary: 'Download flagged file',
|
||||
description: 'Download the untracked/broken file if one exists',
|
||||
history: new HistoryBuilder().added('v3.0.0').alpha('v3.0.0'),
|
||||
})
|
||||
@FileResponse()
|
||||
@Authenticated({ permission: Permission.Maintenance, admin: true })
|
||||
async getIntegrityReportFile(
|
||||
@Param() { id }: UUIDv7ParamDto,
|
||||
@Res() res: Response,
|
||||
@Next() next: NextFunction,
|
||||
): Promise<void> {
|
||||
await sendFile(res, next, () => this.service.getIntegrityReportFile(id), this.logger);
|
||||
}
|
||||
|
||||
@Delete('report/:id')
|
||||
@Endpoint({
|
||||
summary: 'Delete integrity report item',
|
||||
description: 'Delete a given report item and perform corresponding deletion (e.g. trash asset, delete file)',
|
||||
history: new HistoryBuilder().added('v3.0.0').alpha('v3.0.0'),
|
||||
})
|
||||
@Authenticated({ permission: Permission.Maintenance, admin: true })
|
||||
async deleteIntegrityReport(@Auth() auth: AuthDto, @Param() { id }: UUIDv7ParamDto): Promise<void> {
|
||||
await this.service.deleteIntegrityReport(auth.user.id, id);
|
||||
}
|
||||
|
||||
@Get('report/:type/csv')
|
||||
@Endpoint({
|
||||
summary: 'Export integrity report by type as CSV',
|
||||
description: 'Get all integrity report entries for a given type as a CSV',
|
||||
history: new HistoryBuilder().added('v3.0.0').alpha('v3.0.0'),
|
||||
})
|
||||
@FileResponse()
|
||||
@Authenticated({ permission: Permission.Maintenance, admin: true })
|
||||
getIntegrityReportCsv(@Param() { type }: IntegrityReportTypeParamDto, @Res() res: Response): void {
|
||||
res.setHeader('Content-Type', 'text/csv');
|
||||
res.setHeader('Cache-Control', 'private, no-cache, no-transform');
|
||||
res.setHeader('Content-Disposition', `attachment; filename="${encodeURIComponent(`${Date.now()}-${type}.csv`)}"`);
|
||||
|
||||
this.service.getIntegrityReportCsv(type).pipe(res);
|
||||
}
|
||||
}
|
||||
41
server/src/dtos/integrity.dto.ts
Normal file
41
server/src/dtos/integrity.dto.ts
Normal file
|
|
@ -0,0 +1,41 @@
|
|||
import { createZodDto } from 'nestjs-zod';
|
||||
import { IntegrityReport, IntegrityReportSchema } from 'src/enum';
|
||||
import z from 'zod';
|
||||
|
||||
const IntegrityReportSummaryResponseSchema = z
|
||||
.object({
|
||||
[IntegrityReport.ChecksumFail]: z.int().nonnegative(),
|
||||
[IntegrityReport.MissingFile]: z.int().nonnegative(),
|
||||
[IntegrityReport.UntrackedFile]: z.int().nonnegative(),
|
||||
})
|
||||
.meta({ id: 'IntegrityReportSummaryResponseDto' });
|
||||
|
||||
export class IntegrityReportSummaryResponseDto extends createZodDto(IntegrityReportSummaryResponseSchema) {}
|
||||
|
||||
const IntegrityGetReportSchema = z
|
||||
.object({
|
||||
type: IntegrityReportSchema,
|
||||
cursor: z.string().optional().describe('Cursor for pagination'),
|
||||
limit: z.int().positive().default(500).optional().describe('Number of items per page'),
|
||||
})
|
||||
.meta({ id: 'IntegrityGetReportDto' });
|
||||
|
||||
export class IntegrityGetReportDto extends createZodDto(IntegrityGetReportSchema) {}
|
||||
|
||||
const IntegrityDeleteReportSchema = z.object({ type: IntegrityReport }).meta({ id: 'IntegrityDeleteReportDto' });
|
||||
|
||||
export class IntegrityDeleteReportDto extends createZodDto(IntegrityDeleteReportSchema) {}
|
||||
|
||||
const IntegrityReportResponseItemSchema = z.object({
|
||||
id: z.string().describe('Integrity report item id'),
|
||||
type: IntegrityReportSchema,
|
||||
path: z.string().describe('Integrity report item path'),
|
||||
});
|
||||
const IntegrityReportResponseSchema = z
|
||||
.object({
|
||||
items: z.array(IntegrityReportResponseItemSchema),
|
||||
nextCursor: z.string().optional(),
|
||||
})
|
||||
.meta({ id: 'IntegrityReportResponseDto' });
|
||||
|
||||
export class IntegrityReportResponseDto extends createZodDto(IntegrityReportResponseSchema) {}
|
||||
|
|
@ -37,6 +37,7 @@ const QueuesResponseLegacySchema = z
|
|||
[QueueName.Ocr]: QueueResponseLegacySchema,
|
||||
[QueueName.Workflow]: QueueResponseLegacySchema,
|
||||
[QueueName.Editor]: QueueResponseLegacySchema,
|
||||
[QueueName.IntegrityCheck]: QueueResponseLegacySchema,
|
||||
})
|
||||
.meta({ id: 'QueuesResponseLegacyDto' });
|
||||
|
||||
|
|
|
|||
|
|
@ -54,6 +54,30 @@ const DatabaseBackupSchema = z
|
|||
})
|
||||
.meta({ id: 'DatabaseBackupConfig' });
|
||||
|
||||
const SystemConfigIntegrityJobSchema = z
|
||||
.object({
|
||||
enabled: z.boolean().describe('Enabled'),
|
||||
cronExpression: cronExpressionSchema.describe('Cron expression for when the integrity check should run'),
|
||||
})
|
||||
.describe('Integrity job config')
|
||||
.meta({ id: 'SystemConfigIntegrityJob' });
|
||||
|
||||
const SystemConfigIntegrityChecksumJobSchema = SystemConfigIntegrityJobSchema.extend({
|
||||
timeLimit: z.int().nonnegative().describe('How long the integrity checksum job may run for'),
|
||||
percentageLimit: z.int().nonnegative().describe('Percentage limit of the integrity checksum job'),
|
||||
})
|
||||
.describe('Integrity checksum job config')
|
||||
.meta({ id: 'SystemConfigIntegrityChecksumJob' });
|
||||
|
||||
const SystemConfigIntegrityChecksSchema = z
|
||||
.object({
|
||||
missingFiles: SystemConfigIntegrityJobSchema,
|
||||
untrackedFiles: SystemConfigIntegrityJobSchema,
|
||||
checksumFiles: SystemConfigIntegrityChecksumJobSchema,
|
||||
})
|
||||
.describe('Integrity checks config')
|
||||
.meta({ id: 'SystemConfigIntegrityChecks' });
|
||||
|
||||
const SystemConfigBackupsSchema = z.object({ database: DatabaseBackupSchema }).meta({ id: 'SystemConfigBackupsDto' });
|
||||
|
||||
const SystemConfigFFmpegSchema = z
|
||||
|
|
@ -103,6 +127,7 @@ const SystemConfigJobSchema = z
|
|||
ocr: JobSettingsSchema,
|
||||
workflow: JobSettingsSchema,
|
||||
editor: JobSettingsSchema,
|
||||
integrityCheck: JobSettingsSchema,
|
||||
})
|
||||
.meta({ id: 'SystemConfigJobDto' });
|
||||
|
||||
|
|
@ -382,6 +407,7 @@ export const SystemConfigSchema = z
|
|||
templates: SystemConfigTemplatesSchema,
|
||||
server: SystemConfigServerSchema,
|
||||
user: SystemConfigUserSchema,
|
||||
integrityChecks: SystemConfigIntegrityChecksSchema,
|
||||
})
|
||||
.describe('System configuration')
|
||||
.meta({ id: 'SystemConfigDto' });
|
||||
|
|
|
|||
|
|
@ -341,6 +341,7 @@ export enum SystemMetadataKey {
|
|||
SystemFlags = 'system-flags',
|
||||
VersionCheckState = 'version-check-state',
|
||||
License = 'license',
|
||||
IntegrityChecksumCheckpoint = 'integrity-checksum-checkpoint',
|
||||
}
|
||||
|
||||
export enum UserMetadataKey {
|
||||
|
|
@ -398,6 +399,17 @@ export enum SourceType {
|
|||
|
||||
export const SourceTypeSchema = z.enum(SourceType).describe('Face detection source type').meta({ id: 'SourceType' });
|
||||
|
||||
export enum IntegrityReport {
|
||||
UntrackedFile = 'untracked_file',
|
||||
MissingFile = 'missing_file',
|
||||
ChecksumFail = 'checksum_mismatch',
|
||||
}
|
||||
|
||||
export const IntegrityReportSchema = z
|
||||
.enum(IntegrityReport)
|
||||
.describe('Integrity report type')
|
||||
.meta({ id: 'IntegrityReport' });
|
||||
|
||||
export enum ManualJobName {
|
||||
PersonCleanup = 'person-cleanup',
|
||||
TagCleanup = 'tag-cleanup',
|
||||
|
|
@ -405,6 +417,15 @@ export enum ManualJobName {
|
|||
MemoryCleanup = 'memory-cleanup',
|
||||
MemoryCreate = 'memory-create',
|
||||
BackupDatabase = 'backup-database',
|
||||
IntegrityMissingFiles = `integrity-missing-files`,
|
||||
IntegrityUntrackedFiles = `integrity-untracked-files`,
|
||||
IntegrityChecksumFiles = `integrity-checksum-mismatch`,
|
||||
IntegrityMissingFilesRefresh = `integrity-missing-files-refresh`,
|
||||
IntegrityUntrackedFilesRefresh = `integrity-untracked-files-refresh`,
|
||||
IntegrityChecksumFilesRefresh = `integrity-checksum-mismatch-refresh`,
|
||||
IntegrityMissingFilesDeleteAll = `integrity-missing-files-delete-all`,
|
||||
IntegrityUntrackedFilesDeleteAll = `integrity-untracked-files-delete-all`,
|
||||
IntegrityChecksumFilesDeleteAll = `integrity-checksum-mismatch-delete-all`,
|
||||
}
|
||||
|
||||
export const ManualJobNameSchema = z.enum(ManualJobName).describe('Manual job name').meta({ id: 'ManualJobName' });
|
||||
|
|
@ -771,6 +792,7 @@ export enum QueueName {
|
|||
BackupDatabase = 'backupDatabase',
|
||||
Ocr = 'ocr',
|
||||
Workflow = 'workflow',
|
||||
IntegrityCheck = 'integrityCheck',
|
||||
Editor = 'editor',
|
||||
}
|
||||
|
||||
|
|
@ -866,6 +888,18 @@ export enum JobName {
|
|||
|
||||
// Workflow
|
||||
WorkflowAssetTrigger = 'WorkflowAssetTrigger',
|
||||
|
||||
// Integrity
|
||||
IntegrityUntrackedFilesQueueAll = 'IntegrityUntrackedFilesQueueAll',
|
||||
IntegrityUntrackedFiles = 'IntegrityUntrackedFiles',
|
||||
IntegrityUntrackedFilesRefresh = 'IntegrityUntrackedRefresh',
|
||||
IntegrityMissingFilesQueueAll = 'IntegrityMissingFilesQueueAll',
|
||||
IntegrityMissingFiles = 'IntegrityMissingFiles',
|
||||
IntegrityMissingFilesRefresh = 'IntegrityMissingFilesRefresh',
|
||||
IntegrityChecksumFiles = 'IntegrityChecksumFiles',
|
||||
IntegrityChecksumFilesRefresh = 'IntegrityChecksumFilesRefresh',
|
||||
IntegrityDeleteReportType = 'IntegrityDeleteReportType',
|
||||
IntegrityDeleteReports = 'IntegrityDeleteReports',
|
||||
}
|
||||
|
||||
export const JobNameSchema = z.enum(JobName).describe('Job name').meta({ id: 'JobName' });
|
||||
|
|
@ -917,6 +951,7 @@ export enum DatabaseLock {
|
|||
BackupDatabase = 42,
|
||||
MaintenanceOperation = 621,
|
||||
MemoryCreation = 777,
|
||||
IntegrityCheck = 67,
|
||||
VersionCheck = 800,
|
||||
HlsSessionCleanup = 850,
|
||||
}
|
||||
|
|
@ -1131,6 +1166,7 @@ export enum ApiTag {
|
|||
Download = 'Download',
|
||||
Duplicates = 'Duplicates',
|
||||
Faces = 'Faces',
|
||||
Integrity = 'Integrity (admin)',
|
||||
Jobs = 'Jobs',
|
||||
Libraries = 'Libraries',
|
||||
Maintenance = 'Maintenance (admin)',
|
||||
|
|
|
|||
179
server/src/queries/integrity.repository.sql
Normal file
179
server/src/queries/integrity.repository.sql
Normal file
|
|
@ -0,0 +1,179 @@
|
|||
-- NOTE: This file is auto generated by ./sql-generator
|
||||
|
||||
-- IntegrityRepository.getById
|
||||
select
|
||||
"integrity_report".*
|
||||
from
|
||||
"integrity_report"
|
||||
where
|
||||
"id" = $1
|
||||
|
||||
-- IntegrityRepository.getIntegrityReportSummary
|
||||
select
|
||||
"type",
|
||||
count(*) as "count"
|
||||
from
|
||||
"integrity_report"
|
||||
group by
|
||||
"type"
|
||||
|
||||
-- IntegrityRepository.getIntegrityReport
|
||||
select
|
||||
"id",
|
||||
"type",
|
||||
"path",
|
||||
"assetId",
|
||||
"fileAssetId",
|
||||
"createdAt"
|
||||
from
|
||||
"integrity_report"
|
||||
where
|
||||
"type" = $1
|
||||
and "id" <= $2
|
||||
order by
|
||||
"id" desc
|
||||
limit
|
||||
$3
|
||||
|
||||
-- IntegrityRepository.getAssetPathsByPaths
|
||||
select
|
||||
"asset"."originalPath",
|
||||
"asset_file"."path" as "encodedVideoPath"
|
||||
from
|
||||
"asset"
|
||||
left join "asset_file" on "asset"."id" = "asset_file"."assetId"
|
||||
and "asset_file"."type" = $1
|
||||
where
|
||||
(
|
||||
"originalPath" in $2
|
||||
or "asset_file"."path" in $3
|
||||
)
|
||||
|
||||
-- IntegrityRepository.getAssetFilePathsByPaths
|
||||
select
|
||||
"path"
|
||||
from
|
||||
"asset_file"
|
||||
where
|
||||
"path" in $1
|
||||
|
||||
-- IntegrityRepository.getPersonThumbnailPathsByPaths
|
||||
select
|
||||
"person"."thumbnailPath"
|
||||
from
|
||||
"person"
|
||||
where
|
||||
"person"."thumbnailPath" in $1
|
||||
|
||||
-- IntegrityRepository.getAssetCount
|
||||
select
|
||||
count(*) as "count"
|
||||
from
|
||||
"asset"
|
||||
|
||||
-- IntegrityRepository.streamAllAssetPaths
|
||||
select
|
||||
"originalPath",
|
||||
"asset_file"."path" as "encodedVideoPath"
|
||||
from
|
||||
"asset"
|
||||
left join "asset_file" on "asset"."id" = "asset_file"."assetId"
|
||||
and "asset_file"."type" = $1
|
||||
|
||||
-- IntegrityRepository.streamAllAssetFilePaths
|
||||
select
|
||||
"path"
|
||||
from
|
||||
"asset_file"
|
||||
|
||||
-- IntegrityRepository.streamAssetPaths
|
||||
select
|
||||
"allPaths"."path" as "path",
|
||||
"allPaths"."assetId",
|
||||
"allPaths"."fileAssetId",
|
||||
"integrity_report"."id" as "reportId"
|
||||
from
|
||||
(
|
||||
select
|
||||
"asset"."originalPath" as "path",
|
||||
"asset"."id" as "assetId",
|
||||
null::uuid as "fileAssetId"
|
||||
from
|
||||
"asset"
|
||||
where
|
||||
"asset"."deletedAt" is null
|
||||
union all
|
||||
select
|
||||
"path",
|
||||
null::uuid as "assetId",
|
||||
"asset_file"."id" as "fileAssetId"
|
||||
from
|
||||
"asset_file"
|
||||
) as "allPaths"
|
||||
left join "integrity_report" on "integrity_report"."type" = $1
|
||||
and (
|
||||
"integrity_report"."assetId" = "allPaths"."assetId"
|
||||
or "integrity_report"."fileAssetId" = "allPaths"."fileAssetId"
|
||||
)
|
||||
|
||||
-- IntegrityRepository.streamAssetChecksums
|
||||
select
|
||||
"asset"."originalPath",
|
||||
"asset"."checksum",
|
||||
"asset"."createdAt",
|
||||
"asset"."id" as "assetId",
|
||||
"integrity_report"."id" as "reportId"
|
||||
from
|
||||
"asset"
|
||||
left join "integrity_report" on "integrity_report"."assetId" = "asset"."id"
|
||||
and "integrity_report"."type" = $1
|
||||
where
|
||||
"asset"."deletedAt" is null
|
||||
and "createdAt" >= $2
|
||||
and "createdAt" <= $3
|
||||
order by
|
||||
"createdAt" asc
|
||||
|
||||
-- IntegrityRepository.streamIntegrityReports
|
||||
select
|
||||
"id",
|
||||
"type",
|
||||
"path",
|
||||
"assetId",
|
||||
"fileAssetId"
|
||||
from
|
||||
"integrity_report"
|
||||
where
|
||||
"type" = $1
|
||||
order by
|
||||
"createdAt" desc
|
||||
|
||||
-- IntegrityRepository.streamIntegrityReportsWithAssetChecksum
|
||||
select
|
||||
"integrity_report"."id" as "reportId",
|
||||
"integrity_report"."path"
|
||||
from
|
||||
"integrity_report"
|
||||
where
|
||||
"integrity_report"."type" = $1
|
||||
|
||||
-- IntegrityRepository.streamIntegrityReportsByProperty
|
||||
select
|
||||
"id",
|
||||
"path",
|
||||
"assetId",
|
||||
"fileAssetId"
|
||||
from
|
||||
"integrity_report"
|
||||
where
|
||||
"abcdefghi" is not null
|
||||
|
||||
-- IntegrityRepository.deleteById
|
||||
delete from "integrity_report"
|
||||
where
|
||||
"id" = $1
|
||||
|
||||
-- IntegrityRepository.deleteByIds
|
||||
delete from "integrity_report"
|
||||
where
|
||||
"id" in $1
|
||||
|
|
@ -15,6 +15,7 @@ import { DownloadRepository } from 'src/repositories/download.repository';
|
|||
import { DuplicateRepository } from 'src/repositories/duplicate.repository';
|
||||
import { EmailRepository } from 'src/repositories/email.repository';
|
||||
import { EventRepository } from 'src/repositories/event.repository';
|
||||
import { IntegrityRepository } from 'src/repositories/integrity.repository';
|
||||
import { JobRepository } from 'src/repositories/job.repository';
|
||||
import { LibraryRepository } from 'src/repositories/library.repository';
|
||||
import { LoggingRepository } from 'src/repositories/logging.repository';
|
||||
|
|
@ -69,6 +70,7 @@ export const repositories = [
|
|||
DuplicateRepository,
|
||||
EmailRepository,
|
||||
EventRepository,
|
||||
IntegrityRepository,
|
||||
JobRepository,
|
||||
LibraryRepository,
|
||||
LoggingRepository,
|
||||
|
|
|
|||
228
server/src/repositories/integrity.repository.ts
Normal file
228
server/src/repositories/integrity.repository.ts
Normal file
|
|
@ -0,0 +1,228 @@
|
|||
import { Injectable } from '@nestjs/common';
|
||||
import { Insertable, Kysely, sql } from 'kysely';
|
||||
import { InjectKysely } from 'nestjs-kysely';
|
||||
import { DummyValue, GenerateSql } from 'src/decorators';
|
||||
import { AssetFileType, IntegrityReport } from 'src/enum';
|
||||
import { DB } from 'src/schema';
|
||||
import { IntegrityReportTable } from 'src/schema/tables/integrity-report.table';
|
||||
|
||||
export type ReportPaginationOptions = {
|
||||
cursor?: string;
|
||||
limit: number;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class IntegrityRepository {
|
||||
constructor(@InjectKysely() private db: Kysely<DB>) {}
|
||||
|
||||
create(dto: Insertable<IntegrityReportTable> | Insertable<IntegrityReportTable>[]) {
|
||||
return this.db
|
||||
.insertInto('integrity_report')
|
||||
.values(dto)
|
||||
.onConflict((oc) =>
|
||||
oc.columns(['path', 'type']).doUpdateSet({
|
||||
assetId: (eb) => eb.ref('excluded.assetId'),
|
||||
fileAssetId: (eb) => eb.ref('excluded.fileAssetId'),
|
||||
}),
|
||||
)
|
||||
.returningAll()
|
||||
.executeTakeFirstOrThrow();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.STRING] })
|
||||
getById(id: string) {
|
||||
return this.db
|
||||
.selectFrom('integrity_report')
|
||||
.selectAll('integrity_report')
|
||||
.where('id', '=', id)
|
||||
.executeTakeFirstOrThrow();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [] })
|
||||
async getIntegrityReportSummary() {
|
||||
const counts = await this.db
|
||||
.selectFrom('integrity_report')
|
||||
.select(['type', this.db.fn.countAll<number>().as('count')])
|
||||
.groupBy('type')
|
||||
.execute();
|
||||
|
||||
return Object.fromEntries(
|
||||
Object.values(IntegrityReport).map((type) => [type, counts.find((count) => count.type === type)?.count || 0]),
|
||||
) as Record<IntegrityReport, number>;
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [{ cursor: DummyValue.NUMBER, limit: 100 }, DummyValue.STRING] })
|
||||
async getIntegrityReport(pagination: ReportPaginationOptions, type: IntegrityReport) {
|
||||
const items = await this.db
|
||||
.selectFrom('integrity_report')
|
||||
.select(['id', 'type', 'path', 'assetId', 'fileAssetId', 'createdAt'])
|
||||
.where('type', '=', type)
|
||||
.$if(pagination.cursor !== undefined, (eb) => eb.where('id', '<=', pagination.cursor!))
|
||||
.orderBy('id', 'desc')
|
||||
.limit(pagination.limit + 1)
|
||||
.execute();
|
||||
|
||||
return {
|
||||
items: items.slice(0, pagination.limit),
|
||||
nextCursor: items.at(pagination.limit)?.id,
|
||||
};
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.STRING] })
|
||||
getAssetPathsByPaths(paths: string[]) {
|
||||
return this.db
|
||||
.selectFrom('asset')
|
||||
.leftJoin('asset_file', (join) =>
|
||||
join.onRef('asset.id', '=', 'asset_file.assetId').on('asset_file.type', '=', AssetFileType.EncodedVideo),
|
||||
)
|
||||
.select(['asset.originalPath', 'asset_file.path as encodedVideoPath'])
|
||||
.where((eb) => eb.or([eb('originalPath', 'in', paths), eb('asset_file.path', 'in', paths)]))
|
||||
.execute();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.STRING] })
|
||||
getAssetFilePathsByPaths(paths: string[]) {
|
||||
return this.db.selectFrom('asset_file').select('path').where('path', 'in', paths).execute();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.STRING] })
|
||||
getPersonThumbnailPathsByPaths(paths: string[]) {
|
||||
return this.db
|
||||
.selectFrom('person')
|
||||
.select('person.thumbnailPath')
|
||||
.where('person.thumbnailPath', 'in', paths)
|
||||
.execute();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [] })
|
||||
getAssetCount() {
|
||||
return this.db
|
||||
.selectFrom('asset')
|
||||
.select((eb) => eb.fn.countAll<number>().as('count'))
|
||||
.executeTakeFirstOrThrow();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [], stream: true })
|
||||
streamAllAssetPaths() {
|
||||
return this.db
|
||||
.selectFrom('asset')
|
||||
.leftJoin('asset_file', (join) =>
|
||||
join.onRef('asset.id', '=', 'asset_file.assetId').on('asset_file.type', '=', AssetFileType.EncodedVideo),
|
||||
)
|
||||
.select(['originalPath', 'asset_file.path as encodedVideoPath'])
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [], stream: true })
|
||||
streamAllAssetFilePaths() {
|
||||
return this.db.selectFrom('asset_file').select(['path']).stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [], stream: true })
|
||||
streamAssetPaths() {
|
||||
return this.db
|
||||
.selectFrom((eb) =>
|
||||
eb
|
||||
.selectFrom('asset')
|
||||
.where('asset.deletedAt', 'is', null)
|
||||
.select('asset.originalPath as path')
|
||||
.select((eb) => [
|
||||
eb.ref('asset.id').$castTo<string | null>().as('assetId'),
|
||||
sql<string | null>`null::uuid`.as('fileAssetId'),
|
||||
])
|
||||
.unionAll(
|
||||
eb
|
||||
.selectFrom('asset_file')
|
||||
.select(['path'])
|
||||
.select((eb) => [
|
||||
sql<string | null>`null::uuid`.as('assetId'),
|
||||
eb.ref('asset_file.id').$castTo<string | null>().as('fileAssetId'),
|
||||
]),
|
||||
)
|
||||
.as('allPaths'),
|
||||
)
|
||||
.leftJoin('integrity_report', (join) =>
|
||||
join
|
||||
.on('integrity_report.type', '=', IntegrityReport.UntrackedFile)
|
||||
.on((eb) =>
|
||||
eb.or([
|
||||
eb('integrity_report.assetId', '=', eb.ref('allPaths.assetId')),
|
||||
eb('integrity_report.fileAssetId', '=', eb.ref('allPaths.fileAssetId')),
|
||||
]),
|
||||
),
|
||||
)
|
||||
.select(['allPaths.path as path', 'allPaths.assetId', 'allPaths.fileAssetId', 'integrity_report.id as reportId'])
|
||||
.stream() as AsyncIterableIterator<
|
||||
{ path: string; reportId: string | null } & (
|
||||
| { assetId: string; fileAssetId: null }
|
||||
| { assetId: null; fileAssetId: string }
|
||||
)
|
||||
>;
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.DATE, DummyValue.DATE], stream: true })
|
||||
streamAssetChecksums(startMarker?: Date, endMarker?: Date) {
|
||||
return this.db
|
||||
.selectFrom('asset')
|
||||
.where('asset.deletedAt', 'is', null)
|
||||
.leftJoin('integrity_report', (join) =>
|
||||
join
|
||||
.onRef('integrity_report.assetId', '=', 'asset.id')
|
||||
.on('integrity_report.type', '=', IntegrityReport.ChecksumFail),
|
||||
)
|
||||
.select([
|
||||
'asset.originalPath',
|
||||
'asset.checksum',
|
||||
'asset.createdAt',
|
||||
'asset.id as assetId',
|
||||
'integrity_report.id as reportId',
|
||||
])
|
||||
.$if(startMarker !== undefined, (qb) => qb.where('createdAt', '>=', startMarker!))
|
||||
.$if(endMarker !== undefined, (qb) => qb.where('createdAt', '<=', endMarker!))
|
||||
.orderBy('createdAt', 'asc')
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.STRING], stream: true })
|
||||
streamIntegrityReports(type: IntegrityReport) {
|
||||
return this.db
|
||||
.selectFrom('integrity_report')
|
||||
.select(['id', 'type', 'path', 'assetId', 'fileAssetId'])
|
||||
.where('type', '=', type)
|
||||
.orderBy('createdAt', 'desc')
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.STRING], stream: true })
|
||||
streamIntegrityReportsWithAssetChecksum(type: IntegrityReport) {
|
||||
return this.db
|
||||
.selectFrom('integrity_report')
|
||||
.select(['integrity_report.id as reportId', 'integrity_report.path'])
|
||||
.where('integrity_report.type', '=', type)
|
||||
.$if(type === IntegrityReport.ChecksumFail, (eb) =>
|
||||
eb.leftJoin('asset', 'integrity_report.path', 'asset.originalPath').select('asset.checksum'),
|
||||
)
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.STRING], stream: true })
|
||||
streamIntegrityReportsByProperty(property?: 'assetId' | 'fileAssetId', filterType?: IntegrityReport) {
|
||||
return this.db
|
||||
.selectFrom('integrity_report')
|
||||
.select(['id', 'path', 'assetId', 'fileAssetId'])
|
||||
.$if(filterType !== undefined, (eb) => eb.where('type', '=', filterType!))
|
||||
.$if(property === undefined, (eb) => eb.where('assetId', 'is', null).where('fileAssetId', 'is', null))
|
||||
.$if(property !== undefined, (eb) => eb.where(property!, 'is not', null))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.STRING] })
|
||||
deleteById(id: string) {
|
||||
return this.db.deleteFrom('integrity_report').where('id', '=', id).execute();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.STRING] })
|
||||
deleteByIds(ids: string[]) {
|
||||
return this.db.deleteFrom('integrity_report').where('id', 'in', ids).execute();
|
||||
}
|
||||
}
|
||||
|
|
@ -49,6 +49,7 @@ import { AssetOcrTable } from 'src/schema/tables/asset-ocr.table';
|
|||
import { AssetTable } from 'src/schema/tables/asset.table';
|
||||
import { FaceSearchTable } from 'src/schema/tables/face-search.table';
|
||||
import { GeodataPlacesTable } from 'src/schema/tables/geodata-places.table';
|
||||
import { IntegrityReportTable } from 'src/schema/tables/integrity-report.table';
|
||||
import { LibraryTable } from 'src/schema/tables/library.table';
|
||||
import { MemoryAssetAuditTable } from 'src/schema/tables/memory-asset-audit.table';
|
||||
import { MemoryAssetTable } from 'src/schema/tables/memory-asset.table';
|
||||
|
|
@ -115,6 +116,7 @@ export class ImmichDatabase {
|
|||
AssetExifTable,
|
||||
FaceSearchTable,
|
||||
GeodataPlacesTable,
|
||||
IntegrityReportTable,
|
||||
LibraryTable,
|
||||
MemoryTable,
|
||||
MemoryAuditTable,
|
||||
|
|
@ -219,6 +221,8 @@ export interface DB {
|
|||
|
||||
geodata_places: GeodataPlacesTable;
|
||||
|
||||
integrity_report: IntegrityReportTable;
|
||||
|
||||
library: LibraryTable;
|
||||
|
||||
memory: MemoryTable;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,24 @@
|
|||
import { Kysely, sql } from 'kysely';
|
||||
|
||||
export async function up(db: Kysely<any>): Promise<void> {
|
||||
await sql`CREATE TABLE "integrity_report" (
|
||||
"id" uuid NOT NULL DEFAULT immich_uuid_v7(),
|
||||
"type" character varying NOT NULL,
|
||||
"path" character varying NOT NULL,
|
||||
"createdAt" timestamp with time zone NOT NULL DEFAULT now(),
|
||||
"assetId" uuid,
|
||||
"fileAssetId" uuid,
|
||||
CONSTRAINT "integrity_report_assetId_fkey" FOREIGN KEY ("assetId") REFERENCES "asset" ("id") ON UPDATE CASCADE ON DELETE CASCADE,
|
||||
CONSTRAINT "integrity_report_fileAssetId_fkey" FOREIGN KEY ("fileAssetId") REFERENCES "asset_file" ("id") ON UPDATE CASCADE ON DELETE CASCADE,
|
||||
CONSTRAINT "integrity_report_type_path_uq" UNIQUE ("type", "path"),
|
||||
CONSTRAINT "integrity_report_pkey" PRIMARY KEY ("id")
|
||||
);`.execute(db);
|
||||
await sql`CREATE INDEX "integrity_report_assetId_idx" ON "integrity_report" ("assetId");`.execute(db);
|
||||
await sql`CREATE INDEX "integrity_report_fileAssetId_idx" ON "integrity_report" ("fileAssetId");`.execute(db);
|
||||
await sql`CREATE INDEX "asset_createdAt_idx" ON "asset" ("createdAt");`.execute(db);
|
||||
}
|
||||
|
||||
export async function down(db: Kysely<any>): Promise<void> {
|
||||
await sql`DROP TABLE "integrity_report";`.execute(db);
|
||||
await sql`DROP INDEX "asset_createdAt_idx";`.execute(db);
|
||||
}
|
||||
|
|
@ -98,7 +98,7 @@ export class AssetTable {
|
|||
@UpdateDateColumn()
|
||||
updatedAt!: Generated<Timestamp>;
|
||||
|
||||
@CreateDateColumn()
|
||||
@CreateDateColumn({ index: true })
|
||||
createdAt!: Generated<Timestamp>;
|
||||
|
||||
@Column({ index: true })
|
||||
|
|
|
|||
27
server/src/schema/tables/integrity-report.table.ts
Normal file
27
server/src/schema/tables/integrity-report.table.ts
Normal file
|
|
@ -0,0 +1,27 @@
|
|||
import { Column, CreateDateColumn, ForeignKeyColumn, Generated, Table, Timestamp, Unique } from '@immich/sql-tools';
|
||||
import { PrimaryGeneratedUuidV7Column } from 'src/decorators';
|
||||
import { IntegrityReport } from 'src/enum';
|
||||
import { AssetFileTable } from 'src/schema/tables/asset-file.table';
|
||||
import { AssetTable } from 'src/schema/tables/asset.table';
|
||||
|
||||
@Table('integrity_report')
|
||||
@Unique({ columns: ['type', 'path'] })
|
||||
export class IntegrityReportTable {
|
||||
@PrimaryGeneratedUuidV7Column()
|
||||
id!: Generated<string>;
|
||||
|
||||
@Column()
|
||||
type!: IntegrityReport;
|
||||
|
||||
@Column()
|
||||
path!: string;
|
||||
|
||||
@CreateDateColumn()
|
||||
createdAt!: Generated<Timestamp>;
|
||||
|
||||
@ForeignKeyColumn(() => AssetTable, { onDelete: 'CASCADE', onUpdate: 'CASCADE', nullable: true })
|
||||
assetId!: string | null;
|
||||
|
||||
@ForeignKeyColumn(() => AssetFileTable, { onDelete: 'CASCADE', onUpdate: 'CASCADE', nullable: true })
|
||||
fileAssetId!: string | null;
|
||||
}
|
||||
|
|
@ -22,6 +22,7 @@ import { DownloadRepository } from 'src/repositories/download.repository';
|
|||
import { DuplicateRepository } from 'src/repositories/duplicate.repository';
|
||||
import { EmailRepository } from 'src/repositories/email.repository';
|
||||
import { EventRepository } from 'src/repositories/event.repository';
|
||||
import { IntegrityRepository } from 'src/repositories/integrity.repository';
|
||||
import { JobRepository } from 'src/repositories/job.repository';
|
||||
import { LibraryRepository } from 'src/repositories/library.repository';
|
||||
import { LoggingRepository } from 'src/repositories/logging.repository';
|
||||
|
|
@ -81,6 +82,7 @@ export const BASE_SERVICE_DEPENDENCIES = [
|
|||
DuplicateRepository,
|
||||
EmailRepository,
|
||||
EventRepository,
|
||||
IntegrityRepository,
|
||||
JobRepository,
|
||||
LibraryRepository,
|
||||
MachineLearningRepository,
|
||||
|
|
@ -140,6 +142,7 @@ export class BaseService {
|
|||
protected duplicateRepository: DuplicateRepository,
|
||||
protected emailRepository: EmailRepository,
|
||||
protected eventRepository: EventRepository,
|
||||
protected integrityRepository: IntegrityRepository,
|
||||
protected jobRepository: JobRepository,
|
||||
protected libraryRepository: LibraryRepository,
|
||||
protected machineLearningRepository: MachineLearningRepository,
|
||||
|
|
@ -208,6 +211,7 @@ export class BaseService {
|
|||
ctx.duplicateRepository,
|
||||
ctx.emailRepository,
|
||||
ctx.eventRepository,
|
||||
ctx.integrityRepository,
|
||||
ctx.jobRepository,
|
||||
ctx.libraryRepository,
|
||||
ctx.machineLearningRepository,
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ import { DatabaseService } from 'src/services/database.service';
|
|||
import { DownloadService } from 'src/services/download.service';
|
||||
import { DuplicateService } from 'src/services/duplicate.service';
|
||||
import { HlsService } from 'src/services/hls.service';
|
||||
import { IntegrityService } from 'src/services/integrity.service';
|
||||
import { JobService } from 'src/services/job.service';
|
||||
import { LibraryService } from 'src/services/library.service';
|
||||
import { MaintenanceService } from 'src/services/maintenance.service';
|
||||
|
|
@ -63,6 +64,7 @@ export const services = [
|
|||
DatabaseService,
|
||||
DownloadService,
|
||||
DuplicateService,
|
||||
IntegrityService,
|
||||
HlsService,
|
||||
JobService,
|
||||
LibraryService,
|
||||
|
|
|
|||
29
server/src/services/integrity.service.spec.ts
Normal file
29
server/src/services/integrity.service.spec.ts
Normal file
|
|
@ -0,0 +1,29 @@
|
|||
import { IntegrityService } from 'src/services/integrity.service';
|
||||
import { newTestService, ServiceMocks } from 'test/utils';
|
||||
|
||||
describe(IntegrityService.name, () => {
|
||||
let sut: IntegrityService;
|
||||
let mocks: ServiceMocks;
|
||||
|
||||
beforeEach(() => {
|
||||
({ sut, mocks } = newTestService(IntegrityService));
|
||||
});
|
||||
|
||||
it('should work', () => {
|
||||
expect(sut).toBeDefined();
|
||||
});
|
||||
|
||||
describe('handleDeleteAllIntegrityReports', () => {
|
||||
beforeEach(() => {
|
||||
mocks.integrityReport.streamIntegrityReportsByProperty.mockReturnValue((function* () {})() as never);
|
||||
});
|
||||
|
||||
it('should query all property types when no type specified', async () => {
|
||||
await sut.handleDeleteAllIntegrityReports({});
|
||||
|
||||
expect(mocks.integrityReport.streamIntegrityReportsByProperty).toHaveBeenCalledWith(undefined, undefined);
|
||||
expect(mocks.integrityReport.streamIntegrityReportsByProperty).toHaveBeenCalledWith('assetId', undefined);
|
||||
expect(mocks.integrityReport.streamIntegrityReportsByProperty).toHaveBeenCalledWith('fileAssetId', undefined);
|
||||
});
|
||||
});
|
||||
});
|
||||
724
server/src/services/integrity.service.ts
Normal file
724
server/src/services/integrity.service.ts
Normal file
|
|
@ -0,0 +1,724 @@
|
|||
import { Injectable } from '@nestjs/common';
|
||||
import { createHash } from 'node:crypto';
|
||||
import { basename } from 'node:path';
|
||||
import { Readable, Writable } from 'node:stream';
|
||||
import { pipeline } from 'node:stream/promises';
|
||||
import { JOBS_LIBRARY_PAGINATION_SIZE } from 'src/constants';
|
||||
import { StorageCore } from 'src/cores/storage.core';
|
||||
import { OnEvent, OnJob } from 'src/decorators';
|
||||
import {
|
||||
IntegrityGetReportDto,
|
||||
IntegrityReportResponseDto,
|
||||
IntegrityReportSummaryResponseDto,
|
||||
} from 'src/dtos/integrity.dto';
|
||||
import {
|
||||
AssetStatus,
|
||||
CacheControl,
|
||||
DatabaseLock,
|
||||
ImmichWorker,
|
||||
IntegrityReport,
|
||||
JobName,
|
||||
JobStatus,
|
||||
QueueName,
|
||||
StorageFolder,
|
||||
SystemMetadataKey,
|
||||
} from 'src/enum';
|
||||
import { ArgOf } from 'src/repositories/event.repository';
|
||||
import { BaseService } from 'src/services/base.service';
|
||||
import {
|
||||
IIntegrityDeleteReportsJob,
|
||||
IIntegrityDeleteReportTypeJob,
|
||||
IIntegrityJob,
|
||||
IIntegrityMissingFilesJob,
|
||||
IIntegrityPathWithChecksumJob,
|
||||
IIntegrityPathWithReportJob,
|
||||
IIntegrityUntrackedFilesJob,
|
||||
} from 'src/types';
|
||||
import { ImmichFileResponse } from 'src/utils/file';
|
||||
import { handlePromiseError } from 'src/utils/misc';
|
||||
|
||||
/**
|
||||
* Untracked Files:
|
||||
* Files are detected in /data/encoded-video, /data/library, /data/upload
|
||||
* Checked against the asset table
|
||||
* Files are detected in /data/thumbs
|
||||
* Checked against the asset_file table
|
||||
*
|
||||
* * Can perform download or delete of files
|
||||
*
|
||||
* Missing Files:
|
||||
* Paths are queried from asset(originalPath, encodedVideoPath), asset_file(path)
|
||||
* Check whether files exist on disk
|
||||
*
|
||||
* * Reports must include origin (asset or asset_file) & ID for further action
|
||||
* * Can perform trash (asset) or delete (asset_file)
|
||||
*
|
||||
* Checksum Mismatch:
|
||||
* Paths & checksums are queried from asset(originalPath, checksum)
|
||||
* Check whether files match checksum, missing files ignored
|
||||
*
|
||||
* * Reports must include origin (as above) for further action
|
||||
* * Can perform download or trash (asset)
|
||||
*/
|
||||
|
||||
@Injectable()
|
||||
export class IntegrityService extends BaseService {
|
||||
private integrityLock = false;
|
||||
|
||||
@OnEvent({ name: 'ConfigInit', workers: [ImmichWorker.Microservices] })
|
||||
async onConfigInit({
|
||||
newConfig: {
|
||||
integrityChecks: { untrackedFiles, missingFiles, checksumFiles },
|
||||
},
|
||||
}: ArgOf<'ConfigInit'>) {
|
||||
this.integrityLock = await this.databaseRepository.tryLock(DatabaseLock.IntegrityCheck);
|
||||
if (!this.integrityLock) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.cronRepository.create({
|
||||
name: 'integrityUntrackedFiles',
|
||||
expression: untrackedFiles.cronExpression,
|
||||
onTick: () =>
|
||||
handlePromiseError(
|
||||
this.jobRepository.queue({ name: JobName.IntegrityUntrackedFilesQueueAll, data: {} }),
|
||||
this.logger,
|
||||
),
|
||||
start: untrackedFiles.enabled,
|
||||
});
|
||||
|
||||
this.cronRepository.create({
|
||||
name: 'integrityMissingFiles',
|
||||
expression: missingFiles.cronExpression,
|
||||
onTick: () =>
|
||||
handlePromiseError(
|
||||
this.jobRepository.queue({ name: JobName.IntegrityMissingFilesQueueAll, data: {} }),
|
||||
this.logger,
|
||||
),
|
||||
start: missingFiles.enabled,
|
||||
});
|
||||
|
||||
this.cronRepository.create({
|
||||
name: 'integrityChecksumFiles',
|
||||
expression: checksumFiles.cronExpression,
|
||||
onTick: () =>
|
||||
handlePromiseError(this.jobRepository.queue({ name: JobName.IntegrityChecksumFiles, data: {} }), this.logger),
|
||||
start: checksumFiles.enabled,
|
||||
});
|
||||
}
|
||||
|
||||
@OnEvent({ name: 'ConfigUpdate', server: true })
|
||||
onConfigUpdate({
|
||||
newConfig: {
|
||||
integrityChecks: { untrackedFiles, missingFiles, checksumFiles },
|
||||
},
|
||||
}: ArgOf<'ConfigUpdate'>) {
|
||||
if (!this.integrityLock) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.cronRepository.update({
|
||||
name: 'integrityUntrackedFiles',
|
||||
expression: untrackedFiles.cronExpression,
|
||||
start: untrackedFiles.enabled,
|
||||
});
|
||||
|
||||
this.cronRepository.update({
|
||||
name: 'integrityMissingFiles',
|
||||
expression: missingFiles.cronExpression,
|
||||
start: missingFiles.enabled,
|
||||
});
|
||||
|
||||
this.cronRepository.update({
|
||||
name: 'integrityChecksumFiles',
|
||||
expression: checksumFiles.cronExpression,
|
||||
start: checksumFiles.enabled,
|
||||
});
|
||||
}
|
||||
|
||||
getIntegrityReportSummary(): Promise<IntegrityReportSummaryResponseDto> {
|
||||
return this.integrityRepository.getIntegrityReportSummary();
|
||||
}
|
||||
|
||||
getIntegrityReport(dto: IntegrityGetReportDto): Promise<IntegrityReportResponseDto> {
|
||||
return this.integrityRepository.getIntegrityReport({ cursor: dto.cursor, limit: dto.limit ?? 100 }, dto.type);
|
||||
}
|
||||
|
||||
getIntegrityReportCsv(type: IntegrityReport): Readable {
|
||||
const items = this.integrityRepository.streamIntegrityReports(type);
|
||||
|
||||
// very rudimentary csv serialiser
|
||||
async function* generator() {
|
||||
yield 'id,type,assetId,fileAssetId,path\n';
|
||||
|
||||
for await (const item of items) {
|
||||
// no expectation of particularly bad filenames
|
||||
// but they could potentially have a newline or quote character
|
||||
yield `${item.id},${item.type},${item.assetId},${item.fileAssetId},"${item.path.replaceAll('"', '""')}"\n`;
|
||||
}
|
||||
}
|
||||
|
||||
return Readable.from(generator());
|
||||
}
|
||||
|
||||
async getIntegrityReportFile(id: string): Promise<ImmichFileResponse> {
|
||||
const { path } = await this.integrityRepository.getById(id);
|
||||
|
||||
return new ImmichFileResponse({
|
||||
path,
|
||||
fileName: basename(path),
|
||||
contentType: 'application/octet-stream',
|
||||
cacheControl: CacheControl.PrivateWithoutCache,
|
||||
});
|
||||
}
|
||||
|
||||
async deleteIntegrityReport(userId: string, id: string): Promise<void> {
|
||||
const { path, assetId, fileAssetId } = await this.integrityRepository.getById(id);
|
||||
|
||||
if (assetId) {
|
||||
await this.assetRepository.updateAll([assetId], {
|
||||
deletedAt: new Date(),
|
||||
status: AssetStatus.Trashed,
|
||||
});
|
||||
|
||||
await this.eventRepository.emit('AssetTrashAll', {
|
||||
assetIds: [assetId],
|
||||
userId,
|
||||
});
|
||||
|
||||
await this.integrityRepository.deleteById(id);
|
||||
} else if (fileAssetId) {
|
||||
await this.assetRepository.deleteFiles([{ id: fileAssetId }]);
|
||||
} else {
|
||||
await this.storageRepository.unlink(path);
|
||||
await this.integrityRepository.deleteById(id);
|
||||
}
|
||||
}
|
||||
|
||||
private async queueRefreshAllUntrackedFiles() {
|
||||
this.logger.log(`Checking for out of date untracked file reports...`);
|
||||
|
||||
const reports = this.integrityRepository.streamIntegrityReportsWithAssetChecksum(IntegrityReport.UntrackedFile);
|
||||
|
||||
let total = 0;
|
||||
for await (const batchReports of chunk(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
|
||||
await this.jobRepository.queue({
|
||||
name: JobName.IntegrityUntrackedFilesRefresh,
|
||||
data: {
|
||||
items: batchReports,
|
||||
},
|
||||
});
|
||||
|
||||
total += batchReports.length;
|
||||
this.logger.log(`Queued report check of ${batchReports.length} report(s) (${total} so far)`);
|
||||
}
|
||||
}
|
||||
|
||||
@OnJob({ name: JobName.IntegrityUntrackedFilesQueueAll, queue: QueueName.IntegrityCheck })
|
||||
async handleUntrackedFilesQueueAll({ refreshOnly }: IIntegrityJob = {}): Promise<JobStatus> {
|
||||
await this.queueRefreshAllUntrackedFiles();
|
||||
|
||||
if (refreshOnly) {
|
||||
this.logger.log('Refresh complete.');
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
this.logger.log(`Scanning for untracked files...`);
|
||||
|
||||
const assetPaths = this.storageRepository.walk({
|
||||
pathsToCrawl: [StorageFolder.EncodedVideo, StorageFolder.Library, StorageFolder.Upload].map((folder) =>
|
||||
StorageCore.getBaseFolder(folder),
|
||||
),
|
||||
includeHidden: false,
|
||||
take: JOBS_LIBRARY_PAGINATION_SIZE,
|
||||
});
|
||||
|
||||
const assetFilePaths = this.storageRepository.walk({
|
||||
pathsToCrawl: [StorageCore.getBaseFolder(StorageFolder.Thumbnails)],
|
||||
includeHidden: false,
|
||||
take: JOBS_LIBRARY_PAGINATION_SIZE,
|
||||
});
|
||||
|
||||
async function* paths() {
|
||||
for await (const batch of assetPaths) {
|
||||
yield ['asset', batch] as const;
|
||||
}
|
||||
|
||||
for await (const batch of assetFilePaths) {
|
||||
yield ['asset_file', batch] as const;
|
||||
}
|
||||
}
|
||||
|
||||
let total = 0;
|
||||
for await (const [batchType, batchPaths] of paths()) {
|
||||
await this.jobRepository.queue({
|
||||
name: JobName.IntegrityUntrackedFiles,
|
||||
data: {
|
||||
type: batchType,
|
||||
paths: batchPaths,
|
||||
},
|
||||
});
|
||||
|
||||
const count = batchPaths.length;
|
||||
total += count;
|
||||
|
||||
this.logger.log(`Queued untracked check of ${count} file(s) (${total} so far)`);
|
||||
}
|
||||
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
@OnJob({ name: JobName.IntegrityUntrackedFiles, queue: QueueName.IntegrityCheck })
|
||||
async handleUntrackedFiles({ type, paths }: IIntegrityUntrackedFilesJob): Promise<JobStatus> {
|
||||
this.logger.log(`Processing batch of ${paths.length} files to check if they are untracked.`);
|
||||
|
||||
const untrackedFiles = new Set<string>(paths);
|
||||
if (type === 'asset') {
|
||||
const assets = await this.integrityRepository.getAssetPathsByPaths(paths);
|
||||
for (const { originalPath, encodedVideoPath } of assets) {
|
||||
untrackedFiles.delete(originalPath);
|
||||
|
||||
if (encodedVideoPath) {
|
||||
untrackedFiles.delete(encodedVideoPath);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
const assets = await this.integrityRepository.getAssetFilePathsByPaths(paths);
|
||||
for (const { path } of assets) {
|
||||
untrackedFiles.delete(path);
|
||||
}
|
||||
}
|
||||
|
||||
const personThumbnailPaths = await this.integrityRepository.getPersonThumbnailPathsByPaths(paths);
|
||||
for (const { thumbnailPath } of personThumbnailPaths) {
|
||||
untrackedFiles.delete(thumbnailPath);
|
||||
}
|
||||
|
||||
if (untrackedFiles.size > 0) {
|
||||
await this.integrityRepository.create(
|
||||
[...untrackedFiles].map((path) => ({
|
||||
type: IntegrityReport.UntrackedFile,
|
||||
path,
|
||||
})),
|
||||
);
|
||||
}
|
||||
|
||||
this.logger.log(`Processed ${paths.length} and found ${untrackedFiles.size} untracked file(s).`);
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
@OnJob({ name: JobName.IntegrityUntrackedFilesRefresh, queue: QueueName.IntegrityCheck })
|
||||
async handleUntrackedRefresh({ items }: IIntegrityPathWithReportJob): Promise<JobStatus> {
|
||||
this.logger.log(`Processing batch of ${items.length} reports to check if they are out of date.`);
|
||||
|
||||
const results = await Promise.all(
|
||||
items.map(({ reportId, path }) =>
|
||||
this.storageRepository
|
||||
.stat(path)
|
||||
.then(() => void 0)
|
||||
.catch(() => reportId),
|
||||
),
|
||||
);
|
||||
|
||||
const reportIds = results.filter(Boolean) as string[];
|
||||
|
||||
if (reportIds.length > 0) {
|
||||
await this.integrityRepository.deleteByIds(reportIds);
|
||||
}
|
||||
|
||||
this.logger.log(`Processed ${items.length} paths and found ${reportIds.length} report(s) out of date.`);
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
private async queueRefreshAllMissingFiles() {
|
||||
this.logger.log(`Checking for out of date missing file reports...`);
|
||||
|
||||
const reports = this.integrityRepository.streamIntegrityReportsWithAssetChecksum(IntegrityReport.MissingFile);
|
||||
|
||||
let total = 0;
|
||||
for await (const batchReports of chunk(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
|
||||
await this.jobRepository.queue({
|
||||
name: JobName.IntegrityMissingFilesRefresh,
|
||||
data: {
|
||||
items: batchReports,
|
||||
},
|
||||
});
|
||||
|
||||
total += batchReports.length;
|
||||
this.logger.log(`Queued report check of ${batchReports.length} report(s) (${total} so far)`);
|
||||
}
|
||||
|
||||
this.logger.log('Refresh complete.');
|
||||
}
|
||||
|
||||
@OnJob({ name: JobName.IntegrityMissingFilesQueueAll, queue: QueueName.IntegrityCheck })
|
||||
async handleMissingFilesQueueAll({ refreshOnly }: IIntegrityJob = {}): Promise<JobStatus> {
|
||||
if (refreshOnly) {
|
||||
await this.queueRefreshAllMissingFiles();
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
this.logger.log(`Scanning for missing files...`);
|
||||
|
||||
const assetPaths = this.integrityRepository.streamAssetPaths();
|
||||
|
||||
let total = 0;
|
||||
for await (const batchPaths of chunk(assetPaths, JOBS_LIBRARY_PAGINATION_SIZE)) {
|
||||
await this.jobRepository.queue({
|
||||
name: JobName.IntegrityMissingFiles,
|
||||
data: {
|
||||
items: batchPaths,
|
||||
},
|
||||
});
|
||||
|
||||
total += batchPaths.length;
|
||||
this.logger.log(`Queued missing check of ${batchPaths.length} file(s) (${total} so far)`);
|
||||
}
|
||||
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
@OnJob({ name: JobName.IntegrityMissingFiles, queue: QueueName.IntegrityCheck })
|
||||
async handleMissingFiles({ items }: IIntegrityMissingFilesJob): Promise<JobStatus> {
|
||||
this.logger.log(`Processing batch of ${items.length} files to check if they are missing.`);
|
||||
|
||||
const results = await Promise.all(
|
||||
items.map((item) =>
|
||||
this.storageRepository
|
||||
.stat(item.path)
|
||||
.then(() => ({ ...item, exists: true }))
|
||||
.catch(() => ({ ...item, exists: false })),
|
||||
),
|
||||
);
|
||||
|
||||
const outdatedReports = results
|
||||
.filter(({ exists, reportId }) => exists && reportId)
|
||||
.map(({ reportId }) => reportId!);
|
||||
|
||||
if (outdatedReports.length > 0) {
|
||||
await this.integrityRepository.deleteByIds(outdatedReports);
|
||||
}
|
||||
|
||||
const missingFiles = results.filter(({ exists }) => !exists);
|
||||
if (missingFiles.length > 0) {
|
||||
await this.integrityRepository.create(
|
||||
missingFiles.map(({ path, assetId, fileAssetId }) => ({
|
||||
type: IntegrityReport.MissingFile,
|
||||
path,
|
||||
assetId,
|
||||
fileAssetId,
|
||||
})),
|
||||
);
|
||||
}
|
||||
|
||||
this.logger.log(`Processed ${items.length} and found ${missingFiles.length} missing file(s).`);
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
@OnJob({ name: JobName.IntegrityMissingFilesRefresh, queue: QueueName.IntegrityCheck })
|
||||
async handleMissingRefresh({ items: paths }: IIntegrityPathWithReportJob): Promise<JobStatus> {
|
||||
this.logger.log(`Processing batch of ${paths.length} reports to check if they are out of date.`);
|
||||
|
||||
const results = await Promise.all(
|
||||
paths.map(({ reportId, path }) =>
|
||||
this.storageRepository
|
||||
.stat(path)
|
||||
.then(() => reportId)
|
||||
.catch(() => void 0),
|
||||
),
|
||||
);
|
||||
|
||||
const reportIds = results.filter(Boolean) as string[];
|
||||
|
||||
if (reportIds.length > 0) {
|
||||
await this.integrityRepository.deleteByIds(reportIds);
|
||||
}
|
||||
|
||||
this.logger.log(`Processed ${paths.length} paths and found ${reportIds.length} report(s) out of date.`);
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
private async queueRefreshAllChecksumFiles() {
|
||||
this.logger.log(`Checking for out of date checksum file reports...`);
|
||||
|
||||
const reports = this.integrityRepository.streamIntegrityReportsWithAssetChecksum(IntegrityReport.ChecksumFail);
|
||||
|
||||
let total = 0;
|
||||
for await (const batchReports of chunk(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
|
||||
await this.jobRepository.queue({
|
||||
name: JobName.IntegrityChecksumFilesRefresh,
|
||||
data: {
|
||||
items: batchReports.map(({ path, reportId, checksum }) => ({
|
||||
path,
|
||||
reportId,
|
||||
checksum: checksum?.toString('hex'),
|
||||
})),
|
||||
},
|
||||
});
|
||||
|
||||
total += batchReports.length;
|
||||
this.logger.log(`Queued report check of ${batchReports.length} report(s) (${total} so far)`);
|
||||
}
|
||||
|
||||
this.logger.log('Refresh complete.');
|
||||
}
|
||||
|
||||
private async checkAssetChecksum(
|
||||
originalPath: string,
|
||||
checksum: Buffer<ArrayBufferLike>,
|
||||
assetId: string,
|
||||
reportId: string | null,
|
||||
) {
|
||||
const hash = createHash('sha1');
|
||||
|
||||
try {
|
||||
await pipeline([
|
||||
this.storageRepository.createPlainReadStream(originalPath),
|
||||
new Writable({
|
||||
write(chunk, _encoding, callback) {
|
||||
hash.update(chunk);
|
||||
callback();
|
||||
},
|
||||
}),
|
||||
]);
|
||||
|
||||
if (checksum.equals(hash.digest())) {
|
||||
if (reportId) {
|
||||
await this.integrityRepository.deleteById(reportId);
|
||||
}
|
||||
} else {
|
||||
throw new Error('File failed checksum');
|
||||
}
|
||||
} catch (error) {
|
||||
if ((error as { code?: string }).code === 'ENOENT') {
|
||||
if (reportId) {
|
||||
await this.integrityRepository.deleteById(reportId);
|
||||
}
|
||||
|
||||
// missing file; handled by the missing files job
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.warn('Failed to process a file: ' + error);
|
||||
await this.integrityRepository.create({
|
||||
path: originalPath,
|
||||
type: IntegrityReport.ChecksumFail,
|
||||
assetId,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@OnJob({ name: JobName.IntegrityChecksumFiles, queue: QueueName.IntegrityCheck })
|
||||
async handleChecksumFiles({ refreshOnly }: IIntegrityJob = {}): Promise<JobStatus> {
|
||||
if (refreshOnly) {
|
||||
await this.queueRefreshAllChecksumFiles();
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
const {
|
||||
integrityChecks: {
|
||||
checksumFiles: { timeLimit, percentageLimit },
|
||||
},
|
||||
} = await this.getConfig({
|
||||
withCache: true,
|
||||
});
|
||||
|
||||
this.logger.log(
|
||||
`Checking file checksums... (will run for up to ${(timeLimit / (60 * 60 * 1000)).toFixed(2)} hours or until ${(percentageLimit * 100).toFixed(2)}% of assets are processed)`,
|
||||
);
|
||||
|
||||
let processed = 0;
|
||||
const startedAt = Date.now();
|
||||
const { count } = await this.integrityRepository.getAssetCount();
|
||||
const checkpoint = await this.systemMetadataRepository.get(SystemMetadataKey.IntegrityChecksumCheckpoint);
|
||||
|
||||
let startMarker: Date | undefined = checkpoint?.date ? new Date(checkpoint.date) : undefined;
|
||||
let endMarker: Date | undefined;
|
||||
|
||||
const printStats = () => {
|
||||
const averageTime = ((Date.now() - startedAt) / processed).toFixed(2);
|
||||
const completionProgress = ((processed / count) * 100).toFixed(2);
|
||||
|
||||
this.logger.log(
|
||||
`Processed ${processed} files so far... (avg. ${averageTime} ms/asset, ${completionProgress}% of all assets)`,
|
||||
);
|
||||
};
|
||||
|
||||
let lastCreatedAt: Date | undefined;
|
||||
|
||||
finishEarly: do {
|
||||
this.logger.log(
|
||||
`Processing assets in range [${startMarker?.toISOString() ?? 'beginning'}, ${endMarker?.toISOString() ?? 'end'}]`,
|
||||
);
|
||||
|
||||
const assets = this.integrityRepository.streamAssetChecksums(startMarker, endMarker);
|
||||
endMarker = startMarker;
|
||||
startMarker = undefined;
|
||||
|
||||
for await (const { originalPath, checksum, createdAt, assetId, reportId } of assets) {
|
||||
await this.checkAssetChecksum(originalPath, checksum, assetId, reportId);
|
||||
|
||||
processed++;
|
||||
|
||||
if (processed % 100 === 0) {
|
||||
printStats();
|
||||
}
|
||||
|
||||
if (Date.now() > startedAt + timeLimit || processed > count * percentageLimit) {
|
||||
this.logger.log('Reached stop criteria.');
|
||||
lastCreatedAt = createdAt;
|
||||
break finishEarly;
|
||||
}
|
||||
}
|
||||
} while (endMarker);
|
||||
|
||||
await this.systemMetadataRepository.set(SystemMetadataKey.IntegrityChecksumCheckpoint, {
|
||||
date: lastCreatedAt?.toISOString(),
|
||||
});
|
||||
|
||||
printStats();
|
||||
|
||||
if (lastCreatedAt) {
|
||||
this.logger.log(`Finished checksum job, will continue from ${lastCreatedAt.toISOString()}.`);
|
||||
} else {
|
||||
this.logger.log(`Finished checksum job, covered all assets.`);
|
||||
}
|
||||
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
@OnJob({ name: JobName.IntegrityChecksumFilesRefresh, queue: QueueName.IntegrityCheck })
|
||||
async handleChecksumRefresh({ items: paths }: IIntegrityPathWithChecksumJob): Promise<JobStatus> {
|
||||
this.logger.log(`Processing batch of ${paths.length} reports to check if they are out of date.`);
|
||||
|
||||
const results = await Promise.all(
|
||||
paths.map(async ({ reportId, path, checksum }) => {
|
||||
if (!checksum) {
|
||||
return reportId;
|
||||
}
|
||||
|
||||
const hash = createHash('sha1');
|
||||
|
||||
try {
|
||||
await pipeline([
|
||||
this.storageRepository.createPlainReadStream(path),
|
||||
new Writable({
|
||||
write(chunk, _encoding, callback) {
|
||||
hash.update(chunk);
|
||||
callback();
|
||||
},
|
||||
}),
|
||||
]);
|
||||
} catch (error) {
|
||||
if ((error as { code?: string }).code === 'ENOENT') {
|
||||
return reportId;
|
||||
}
|
||||
}
|
||||
|
||||
if (Buffer.from(checksum, 'hex').equals(hash.digest())) {
|
||||
return reportId;
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
const reportIds = results.filter(Boolean) as string[];
|
||||
|
||||
if (reportIds.length > 0) {
|
||||
await this.integrityRepository.deleteByIds(reportIds);
|
||||
}
|
||||
|
||||
this.logger.log(`Processed ${paths.length} paths and found ${reportIds.length} report(s) out of date.`);
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
@OnJob({ name: JobName.IntegrityDeleteReportType, queue: QueueName.IntegrityCheck })
|
||||
async handleDeleteAllIntegrityReports({ type }: IIntegrityDeleteReportTypeJob): Promise<JobStatus> {
|
||||
this.logger.log(`Deleting all entries for ${type ?? 'all types of'} integrity report`);
|
||||
|
||||
let properties;
|
||||
switch (type) {
|
||||
case IntegrityReport.ChecksumFail: {
|
||||
properties = ['assetId'] as const;
|
||||
break;
|
||||
}
|
||||
case IntegrityReport.MissingFile: {
|
||||
properties = ['assetId', 'fileAssetId'] as const;
|
||||
break;
|
||||
}
|
||||
case IntegrityReport.UntrackedFile: {
|
||||
properties = [void 0] as const;
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
properties = [void 0, 'assetId', 'fileAssetId'] as const;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
for (const property of properties) {
|
||||
const reports = this.integrityRepository.streamIntegrityReportsByProperty(property, type);
|
||||
for await (const batch of chunk(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
|
||||
await this.jobRepository.queue({
|
||||
name: JobName.IntegrityDeleteReports,
|
||||
data: {
|
||||
reports: batch,
|
||||
},
|
||||
});
|
||||
|
||||
this.logger.log(`Queued ${batch.length} reports to delete.`);
|
||||
}
|
||||
}
|
||||
|
||||
return JobStatus.Success;
|
||||
}
|
||||
|
||||
@OnJob({ name: JobName.IntegrityDeleteReports, queue: QueueName.IntegrityCheck })
|
||||
async handleDeleteIntegrityReports({ reports }: IIntegrityDeleteReportsJob): Promise<JobStatus> {
|
||||
const byAsset = reports.filter((report) => report.assetId);
|
||||
const byFileAsset = reports.filter((report) => report.fileAssetId);
|
||||
const byPath = reports.filter((report) => !report.assetId && !report.fileAssetId);
|
||||
|
||||
if (byAsset.length > 0) {
|
||||
const ids = byAsset.map(({ assetId }) => assetId!);
|
||||
await this.assetRepository.updateAll(ids, {
|
||||
deletedAt: new Date(),
|
||||
status: AssetStatus.Trashed,
|
||||
});
|
||||
|
||||
await this.eventRepository.emit('AssetTrashAll', {
|
||||
assetIds: ids,
|
||||
userId: '', // we don't notify any users currently
|
||||
});
|
||||
|
||||
await this.integrityRepository.deleteByIds(byAsset.map(({ id }) => id));
|
||||
}
|
||||
|
||||
if (byFileAsset.length > 0) {
|
||||
await this.assetRepository.deleteFiles(byFileAsset.map(({ fileAssetId }) => ({ id: fileAssetId! })));
|
||||
}
|
||||
|
||||
if (byPath.length > 0) {
|
||||
await Promise.all(byPath.map(({ path }) => this.storageRepository.unlink(path).catch(() => void 0)));
|
||||
await this.integrityRepository.deleteByIds(byPath.map(({ id }) => id));
|
||||
}
|
||||
|
||||
this.logger.log(`Deleted ${reports.length} reports.`);
|
||||
return JobStatus.Success;
|
||||
}
|
||||
}
|
||||
|
||||
async function* chunk<T>(generator: AsyncIterableIterator<T>, n: number) {
|
||||
let chunk: T[] = [];
|
||||
for await (const item of generator) {
|
||||
chunk.push(item);
|
||||
|
||||
if (chunk.length === n) {
|
||||
yield chunk;
|
||||
chunk = [];
|
||||
}
|
||||
}
|
||||
|
||||
if (chunk.length > 0) {
|
||||
yield chunk;
|
||||
}
|
||||
}
|
||||
|
|
@ -2,7 +2,7 @@ import { BadRequestException, Injectable } from '@nestjs/common';
|
|||
import { OnEvent } from 'src/decorators';
|
||||
import { mapAsset } from 'src/dtos/asset-response.dto';
|
||||
import { JobCreateDto } from 'src/dtos/job.dto';
|
||||
import { AssetType, AssetVisibility, JobName, JobStatus, ManualJobName } from 'src/enum';
|
||||
import { AssetType, AssetVisibility, IntegrityReport, JobName, JobStatus, ManualJobName } from 'src/enum';
|
||||
import { ArgsOf } from 'src/repositories/event.repository';
|
||||
import { BaseService } from 'src/services/base.service';
|
||||
import { JobItem } from 'src/types';
|
||||
|
|
@ -34,6 +34,42 @@ const asJobItem = (dto: JobCreateDto): JobItem => {
|
|||
return { name: JobName.DatabaseBackup };
|
||||
}
|
||||
|
||||
case ManualJobName.IntegrityMissingFiles: {
|
||||
return { name: JobName.IntegrityMissingFilesQueueAll };
|
||||
}
|
||||
|
||||
case ManualJobName.IntegrityUntrackedFiles: {
|
||||
return { name: JobName.IntegrityUntrackedFilesQueueAll };
|
||||
}
|
||||
|
||||
case ManualJobName.IntegrityChecksumFiles: {
|
||||
return { name: JobName.IntegrityChecksumFiles };
|
||||
}
|
||||
|
||||
case ManualJobName.IntegrityMissingFilesRefresh: {
|
||||
return { name: JobName.IntegrityMissingFilesQueueAll, data: { refreshOnly: true } };
|
||||
}
|
||||
|
||||
case ManualJobName.IntegrityUntrackedFilesRefresh: {
|
||||
return { name: JobName.IntegrityUntrackedFilesQueueAll, data: { refreshOnly: true } };
|
||||
}
|
||||
|
||||
case ManualJobName.IntegrityChecksumFilesRefresh: {
|
||||
return { name: JobName.IntegrityChecksumFiles, data: { refreshOnly: true } };
|
||||
}
|
||||
|
||||
case ManualJobName.IntegrityMissingFilesDeleteAll: {
|
||||
return { name: JobName.IntegrityDeleteReportType, data: { type: IntegrityReport.MissingFile } };
|
||||
}
|
||||
|
||||
case ManualJobName.IntegrityUntrackedFilesDeleteAll: {
|
||||
return { name: JobName.IntegrityDeleteReportType, data: { type: IntegrityReport.UntrackedFile } };
|
||||
}
|
||||
|
||||
case ManualJobName.IntegrityChecksumFilesDeleteAll: {
|
||||
return { name: JobName.IntegrityDeleteReportType, data: { type: IntegrityReport.ChecksumFail } };
|
||||
}
|
||||
|
||||
default: {
|
||||
throw new BadRequestException('Invalid job name');
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ describe(QueueService.name, () => {
|
|||
it('should update concurrency', () => {
|
||||
sut.onConfigUpdate({ newConfig: defaults, oldConfig: {} as SystemConfig });
|
||||
|
||||
expect(mocks.job.setConcurrency).toHaveBeenCalledTimes(18);
|
||||
expect(mocks.job.setConcurrency).toHaveBeenCalledTimes(19);
|
||||
expect(mocks.job.setConcurrency).toHaveBeenNthCalledWith(5, QueueName.FacialRecognition, 1);
|
||||
expect(mocks.job.setConcurrency).toHaveBeenNthCalledWith(7, QueueName.DuplicateDetection, 1);
|
||||
expect(mocks.job.setConcurrency).toHaveBeenNthCalledWith(8, QueueName.BackgroundTask, 5);
|
||||
|
|
@ -77,6 +77,7 @@ describe(QueueService.name, () => {
|
|||
[QueueName.BackupDatabase]: expected,
|
||||
[QueueName.Ocr]: expected,
|
||||
[QueueName.Workflow]: expected,
|
||||
[QueueName.IntegrityCheck]: expected,
|
||||
[QueueName.Editor]: expected,
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -42,6 +42,7 @@ const updatedConfig = Object.freeze<SystemConfig>({
|
|||
[QueueName.Notification]: { concurrency: 5 },
|
||||
[QueueName.Ocr]: { concurrency: 1 },
|
||||
[QueueName.Workflow]: { concurrency: 5 },
|
||||
[QueueName.IntegrityCheck]: { concurrency: 1 },
|
||||
[QueueName.Editor]: { concurrency: 2 },
|
||||
},
|
||||
backup: {
|
||||
|
|
@ -77,6 +78,22 @@ const updatedConfig = Object.freeze<SystemConfig>({
|
|||
enabled: false,
|
||||
},
|
||||
},
|
||||
integrityChecks: {
|
||||
untrackedFiles: {
|
||||
enabled: true,
|
||||
cronExpression: '0 03 * * *',
|
||||
},
|
||||
missingFiles: {
|
||||
enabled: true,
|
||||
cronExpression: '0 03 * * *',
|
||||
},
|
||||
checksumFiles: {
|
||||
enabled: true,
|
||||
cronExpression: '0 03 * * *',
|
||||
timeLimit: 60 * 60 * 1000,
|
||||
percentageLimit: 1,
|
||||
},
|
||||
},
|
||||
logging: {
|
||||
enabled: true,
|
||||
level: LogLevel.Log,
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ import {
|
|||
H264Profile,
|
||||
HevcProfile,
|
||||
ImageFormat,
|
||||
IntegrityReport,
|
||||
JobName,
|
||||
MemoryType,
|
||||
QueueName,
|
||||
|
|
@ -311,6 +312,43 @@ export type IWorkflowJob<T extends WorkflowType = WorkflowType> = {
|
|||
type: T;
|
||||
};
|
||||
|
||||
export interface IIntegrityJob {
|
||||
refreshOnly?: boolean;
|
||||
}
|
||||
|
||||
export interface IIntegrityDeleteReportTypeJob {
|
||||
type?: IntegrityReport;
|
||||
}
|
||||
|
||||
export interface IIntegrityDeleteReportsJob {
|
||||
reports: {
|
||||
id: string;
|
||||
assetId: string | null;
|
||||
fileAssetId: string | null;
|
||||
path: string;
|
||||
}[];
|
||||
}
|
||||
|
||||
export interface IIntegrityUntrackedFilesJob {
|
||||
type: 'asset' | 'asset_file';
|
||||
paths: string[];
|
||||
}
|
||||
|
||||
export interface IIntegrityMissingFilesJob {
|
||||
items: ({ path: string; reportId: string | null } & (
|
||||
| { assetId: string; fileAssetId: null }
|
||||
| { assetId: null; fileAssetId: string }
|
||||
))[];
|
||||
}
|
||||
|
||||
export interface IIntegrityPathWithReportJob {
|
||||
items: { path: string; reportId: string | null }[];
|
||||
}
|
||||
|
||||
export interface IIntegrityPathWithChecksumJob {
|
||||
items: { path: string; reportId: string | null; checksum?: string | null }[];
|
||||
}
|
||||
|
||||
export interface JobCounts {
|
||||
active: number;
|
||||
completed: number;
|
||||
|
|
@ -422,6 +460,18 @@ export type JobItem =
|
|||
// Workflow
|
||||
| { name: JobName.WorkflowAssetTrigger; data: { workflowId: string; assetId: string } }
|
||||
|
||||
// Integrity
|
||||
| { name: JobName.IntegrityUntrackedFilesQueueAll; data?: IIntegrityJob }
|
||||
| { name: JobName.IntegrityUntrackedFiles; data: IIntegrityUntrackedFilesJob }
|
||||
| { name: JobName.IntegrityUntrackedFilesRefresh; data: IIntegrityPathWithReportJob }
|
||||
| { name: JobName.IntegrityMissingFilesQueueAll; data?: IIntegrityJob }
|
||||
| { name: JobName.IntegrityMissingFiles; data: IIntegrityPathWithReportJob }
|
||||
| { name: JobName.IntegrityMissingFilesRefresh; data: IIntegrityPathWithReportJob }
|
||||
| { name: JobName.IntegrityChecksumFiles; data?: IIntegrityJob }
|
||||
| { name: JobName.IntegrityChecksumFilesRefresh; data?: IIntegrityPathWithChecksumJob }
|
||||
| { name: JobName.IntegrityDeleteReportType; data: IIntegrityDeleteReportTypeJob }
|
||||
| { name: JobName.IntegrityDeleteReports; data: IIntegrityDeleteReportsJob }
|
||||
|
||||
// Editor
|
||||
| { name: JobName.AssetEditThumbnailGeneration; data: IEntityJob };
|
||||
|
||||
|
|
@ -522,6 +572,7 @@ export interface SystemMetadata extends Record<SystemMetadataKey, Record<string,
|
|||
[SystemMetadataKey.SystemFlags]: DeepPartial<SystemFlags>;
|
||||
[SystemMetadataKey.VersionCheckState]: VersionCheckMetadata;
|
||||
[SystemMetadataKey.MemoriesState]: MemoriesState;
|
||||
[SystemMetadataKey.IntegrityChecksumCheckpoint]: { date?: string };
|
||||
}
|
||||
|
||||
export type UserPreferences = {
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import { ArgumentMetadata, FileValidator, Injectable, ParseUUIDPipe } from '@nestjs/common';
|
||||
import { createZodDto } from 'nestjs-zod';
|
||||
import sanitize from 'sanitize-filename';
|
||||
import { IntegrityReportSchema } from 'src/enum';
|
||||
import { isIP, isIPRange } from 'validator';
|
||||
import z from 'zod';
|
||||
|
||||
|
|
@ -110,6 +111,12 @@ const UUIDParamSchema = z.object({
|
|||
|
||||
export class UUIDParamDto extends createZodDto(UUIDParamSchema) {}
|
||||
|
||||
const UUIDv7ParamSchema = z.object({
|
||||
id: z.uuidv7(),
|
||||
});
|
||||
|
||||
export class UUIDv7ParamDto extends createZodDto(UUIDv7ParamSchema) {}
|
||||
|
||||
const UUIDAssetIDParamSchema = z.object({
|
||||
id: z.uuidv4(),
|
||||
assetId: z.uuidv4(),
|
||||
|
|
@ -125,6 +132,10 @@ const FilenameParamSchema = z.object({
|
|||
|
||||
export class FilenameParamDto extends createZodDto(FilenameParamSchema) {}
|
||||
|
||||
const IntegrityReportParamSchema = z.object({ type: IntegrityReportSchema }).meta({ id: 'IntegrityReportDto' });
|
||||
|
||||
export class IntegrityReportTypeParamDto extends createZodDto(IntegrityReportParamSchema) {}
|
||||
|
||||
/**
|
||||
* Unified email validation
|
||||
* Converts email strings to lowercase and validates against HTML5 email regex
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ import { CryptoRepository } from 'src/repositories/crypto.repository';
|
|||
import { DatabaseRepository } from 'src/repositories/database.repository';
|
||||
import { EmailRepository } from 'src/repositories/email.repository';
|
||||
import { EventRepository } from 'src/repositories/event.repository';
|
||||
import { IntegrityRepository } from 'src/repositories/integrity.repository';
|
||||
import { JobRepository } from 'src/repositories/job.repository';
|
||||
import { LoggingRepository } from 'src/repositories/logging.repository';
|
||||
import { MachineLearningRepository } from 'src/repositories/machine-learning.repository';
|
||||
|
|
@ -415,6 +416,7 @@ const newRealRepository = <T>(key: ClassConstructor<T>, db: Kysely<DB>): T => {
|
|||
case AssetRepository:
|
||||
case AssetEditRepository:
|
||||
case AssetJobRepository:
|
||||
case IntegrityRepository:
|
||||
case MemoryRepository:
|
||||
case NotificationRepository:
|
||||
case OcrRepository:
|
||||
|
|
@ -483,6 +485,7 @@ const newMockRepository = <T>(key: ClassConstructor<T>) => {
|
|||
case ConfigRepository:
|
||||
case CryptoRepository:
|
||||
case MemoryRepository:
|
||||
case IntegrityRepository:
|
||||
case NotificationRepository:
|
||||
case OcrRepository:
|
||||
case PartnerRepository:
|
||||
|
|
|
|||
1000
server/test/medium/specs/services/integrity.service.spec.ts
Normal file
1000
server/test/medium/specs/services/integrity.service.spec.ts
Normal file
File diff suppressed because it is too large
Load diff
|
|
@ -48,8 +48,8 @@ export const newStorageRepositoryMock = (): Mocked<RepositoryInterface<StorageRe
|
|||
|
||||
return {
|
||||
createZipStream: vitest.fn(),
|
||||
createReadStream: vitest.fn(),
|
||||
createPlainReadStream: vitest.fn(),
|
||||
createReadStream: vitest.fn(),
|
||||
createGzip: vitest.fn(),
|
||||
createGunzip: vitest.fn(),
|
||||
readFile: vitest.fn(),
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ import { DownloadRepository } from 'src/repositories/download.repository';
|
|||
import { DuplicateRepository } from 'src/repositories/duplicate.repository';
|
||||
import { EmailRepository } from 'src/repositories/email.repository';
|
||||
import { EventRepository } from 'src/repositories/event.repository';
|
||||
import { IntegrityRepository } from 'src/repositories/integrity.repository';
|
||||
import { JobRepository } from 'src/repositories/job.repository';
|
||||
import { LibraryRepository } from 'src/repositories/library.repository';
|
||||
import { LoggingRepository } from 'src/repositories/logging.repository';
|
||||
|
|
@ -234,6 +235,7 @@ export type ServiceOverrides = {
|
|||
duplicateRepository: DuplicateRepository;
|
||||
email: EmailRepository;
|
||||
event: EventRepository;
|
||||
integrityReport: IntegrityRepository;
|
||||
job: JobRepository;
|
||||
library: LibraryRepository;
|
||||
logger: LoggingRepository;
|
||||
|
|
@ -316,6 +318,7 @@ export const getMocks = () => {
|
|||
email: automock(EmailRepository, { args: [loggerMock] }),
|
||||
// eslint-disable-next-line no-sparse-arrays
|
||||
event: automock(EventRepository, { args: [, , loggerMock], strict: false }),
|
||||
integrityReport: automock(IntegrityRepository, { strict: false }),
|
||||
job: newJobRepositoryMock(),
|
||||
apiKey: automock(ApiKeyRepository),
|
||||
library: automock(LibraryRepository, { strict: false }),
|
||||
|
|
@ -385,6 +388,7 @@ export const newTestService = <T extends BaseService>(
|
|||
overrides.duplicateRepository || (mocks.duplicateRepository as As<DuplicateRepository>),
|
||||
overrides.email || (mocks.email as As<EmailRepository>),
|
||||
overrides.event || (mocks.event as As<EventRepository>),
|
||||
overrides.integrityReport || (mocks.integrityReport as As<IntegrityRepository>),
|
||||
overrides.job || (mocks.job as As<JobRepository>),
|
||||
overrides.library || (mocks.library as As<LibraryRepository>),
|
||||
overrides.machineLearning || (mocks.machineLearning as As<MachineLearningRepository>),
|
||||
|
|
@ -540,7 +544,44 @@ export const mockDuplex =
|
|||
return duplex;
|
||||
};
|
||||
|
||||
export async function* makeStream<T>(items: T[] = []): AsyncIterableIterator<T> {
|
||||
export const mockFork = vitest.fn((exitCode: number, stdout: string, stderr: string, error?: unknown) => {
|
||||
const stdoutStream = new Readable({
|
||||
read() {
|
||||
this.push(stdout); // write mock data to stdout
|
||||
this.push(null); // end stream
|
||||
},
|
||||
});
|
||||
|
||||
return {
|
||||
stdout: stdoutStream,
|
||||
stderr: new Readable({
|
||||
read() {
|
||||
this.push(stderr); // write mock data to stderr
|
||||
this.push(null); // end stream
|
||||
},
|
||||
}),
|
||||
stdin: new Writable({
|
||||
write(chunk, encoding, callback) {
|
||||
callback();
|
||||
},
|
||||
}),
|
||||
exitCode,
|
||||
on: vitest.fn((event, callback: any) => {
|
||||
if (event === 'close') {
|
||||
stdoutStream.once('end', () => callback(0));
|
||||
}
|
||||
if (event === 'error' && error) {
|
||||
stdoutStream.once('end', () => callback(error));
|
||||
}
|
||||
if (event === 'exit') {
|
||||
stdoutStream.once('end', () => callback(exitCode));
|
||||
}
|
||||
}),
|
||||
kill: vitest.fn(),
|
||||
} as unknown as ChildProcessWithoutNullStreams;
|
||||
});
|
||||
|
||||
export async function* makeStream<T>(items: T[] = []): AsyncGenerator<T> {
|
||||
for (const item of items) {
|
||||
await Promise.resolve();
|
||||
yield item;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue