Merge branch 'main' into feat/memories-view

This commit is contained in:
Timon
2026-06-03 09:46:49 +02:00
committed by GitHub
350 changed files with 21700 additions and 5738 deletions
+9
View File
@@ -1,4 +1,5 @@
import { CronExpression } from '@nestjs/schedule';
import { ReleaseChannel } from 'src/dtos/system-config.dto';
import {
AudioCodec,
Colorspace,
@@ -45,6 +46,9 @@ export type SystemConfig = {
accel: TranscodeHardwareAcceleration;
accelDecode: boolean;
tonemap: ToneMapping;
realtime: {
enabled: boolean;
};
};
job: Record<ConcurrentQueueName, { concurrency: number }>;
logging: {
@@ -135,6 +139,7 @@ export type SystemConfig = {
};
newVersionCheck: {
enabled: boolean;
channel: ReleaseChannel;
};
nightlyTasks: {
startTime: string;
@@ -224,6 +229,9 @@ export const defaults = Object.freeze<SystemConfig>({
tonemap: ToneMapping.Hable,
accel: TranscodeHardwareAcceleration.Disabled,
accelDecode: true,
realtime: {
enabled: false,
},
},
job: {
[QueueName.BackgroundTask]: { concurrency: 5 },
@@ -344,6 +352,7 @@ export const defaults = Object.freeze<SystemConfig>({
},
newVersionCheck: {
enabled: true,
channel: ReleaseChannel.Stable,
},
nightlyTasks: {
startTime: '00:00',
+38 -1
View File
@@ -1,7 +1,15 @@
import { readFileSync } from 'node:fs';
import { dirname, join } from 'node:path';
import { SemVer } from 'semver';
import { ApiTag, AudioCodec, DatabaseExtension, ExifOrientation, VectorIndex } from 'src/enum';
import {
ApiTag,
AudioCodec,
DatabaseExtension,
ExifOrientation,
TranscodeHardwareAcceleration,
VectorIndex,
VideoCodec,
} from 'src/enum';
export const IMMICH_SERVER_START = 'Immich Server is listening';
@@ -202,3 +210,32 @@ export const AUDIO_ENCODER: Record<AudioCodec, string> = {
[AudioCodec.Opus]: 'libopus',
[AudioCodec.PcmS16le]: 'pcm_s16le',
};
export const SUPPORTED_HWA_CODECS: Record<TranscodeHardwareAcceleration, VideoCodec[]> = {
[TranscodeHardwareAcceleration.Nvenc]: [VideoCodec.H264, VideoCodec.Hevc, VideoCodec.Av1],
[TranscodeHardwareAcceleration.Qsv]: [VideoCodec.H264, VideoCodec.Hevc, VideoCodec.Vp9, VideoCodec.Av1],
[TranscodeHardwareAcceleration.Vaapi]: [VideoCodec.H264, VideoCodec.Hevc, VideoCodec.Vp9, VideoCodec.Av1],
[TranscodeHardwareAcceleration.Rkmpp]: [VideoCodec.H264, VideoCodec.Hevc],
[TranscodeHardwareAcceleration.Disabled]: [VideoCodec.H264, VideoCodec.Hevc, VideoCodec.Vp9, VideoCodec.Av1],
};
export const HLS_BACKPRESSURE_PAUSE_SEGMENTS = 30;
export const HLS_BACKPRESSURE_RESUME_SEGMENTS = 15;
export const HLS_CLEANUP_INTERVAL_MS = 60 * 1000;
export const HLS_INACTIVITY_TIMEOUT_MS = 5 * 60 * 1000;
export const HLS_LEASE_DURATION_MS = 30 * 60 * 1000;
export const HLS_PLAYLIST_CONTENT_TYPE = 'application/vnd.apple.mpegurl';
export const HLS_SEGMENT_DURATION = 2;
export const HLS_SEGMENT_FILENAME_REGEX = /^seg_(\d+)\.m4s$/;
export const HLS_VARIANTS = [
{ resolution: 480, codec: VideoCodec.Av1, bitrate: 1_000_000, codecString: 'av01.0.04M.08' },
{ resolution: 480, codec: VideoCodec.Hevc, bitrate: 1_200_000, codecString: 'hvc1.1.6.L90.B0' },
{ resolution: 480, codec: VideoCodec.H264, bitrate: 2_500_000, codecString: 'avc1.64001e' },
{ resolution: 720, codec: VideoCodec.Av1, bitrate: 2_000_000, codecString: 'av01.0.08M.08' },
{ resolution: 720, codec: VideoCodec.Hevc, bitrate: 2_500_000, codecString: 'hvc1.1.6.L93.B0' },
{ resolution: 720, codec: VideoCodec.H264, bitrate: 5_000_000, codecString: 'avc1.64001f' },
{ resolution: 1080, codec: VideoCodec.Av1, bitrate: 4_000_000, codecString: 'av01.0.09M.08' },
{ resolution: 1080, codec: VideoCodec.Hevc, bitrate: 4_500_000, codecString: 'hvc1.1.6.L120.B0' },
{ resolution: 1080, codec: VideoCodec.H264, bitrate: 8_000_000, codecString: 'avc1.640028' },
];
export const HLS_VERSION = 7;
+2
View File
@@ -35,6 +35,7 @@ import { TimelineController } from 'src/controllers/timeline.controller';
import { TrashController } from 'src/controllers/trash.controller';
import { UserAdminController } from 'src/controllers/user-admin.controller';
import { UserController } from 'src/controllers/user.controller';
import { VideoStreamController } from 'src/controllers/video-stream.controller';
import { ViewController } from 'src/controllers/view.controller';
import { WorkflowController } from 'src/controllers/workflow.controller';
@@ -76,6 +77,7 @@ export const controllers = [
TrashController,
UserAdminController,
UserController,
VideoStreamController,
ViewController,
WorkflowController,
];
@@ -0,0 +1,79 @@
import { Controller, Delete, Get, Header, HttpCode, HttpStatus, Next, Param, Res } from '@nestjs/common';
import { ApiProduces, ApiTags } from '@nestjs/swagger';
import { NextFunction, Response } from 'express';
import { HLS_PLAYLIST_CONTENT_TYPE } from 'src/constants';
import { Endpoint, HistoryBuilder } from 'src/decorators';
import { AuthDto } from 'src/dtos/auth.dto';
import { HlsSegmentParamDto, HlsSessionParamDto, HlsVariantParamDto } from 'src/dtos/streaming.dto';
import { ApiTag, Permission, RouteKey } from 'src/enum';
import { Auth, Authenticated, FileResponse } from 'src/middleware/auth.guard';
import { LoggingRepository } from 'src/repositories/logging.repository';
import { HlsService } from 'src/services/hls.service';
import { sendFile } from 'src/utils/file';
import { UUIDParamDto } from 'src/validation';
@ApiTags(ApiTag.Assets)
@Controller(RouteKey.Asset)
export class VideoStreamController {
constructor(
private logger: LoggingRepository,
private service: HlsService,
) {}
@Get(':id/video/stream/main.m3u8')
@Authenticated({ permission: Permission.AssetView, sharedLink: true })
@Header('Cache-Control', 'no-cache')
@Header('Content-Type', HLS_PLAYLIST_CONTENT_TYPE)
@ApiProduces(HLS_PLAYLIST_CONTENT_TYPE)
@Endpoint({
summary: 'Get HLS main playlist',
description: 'Returns an HLS main playlist with all available variants for the asset.',
history: new HistoryBuilder().added('v3').alpha('v3'),
})
getMainPlaylist(@Auth() auth: AuthDto, @Param() { id }: UUIDParamDto) {
return this.service.getMainPlaylist(auth, id);
}
@Get(':id/video/stream/:sessionId/:variantIndex/playlist.m3u8')
@Authenticated({ permission: Permission.AssetView, sharedLink: true })
@Header('Cache-Control', 'no-cache')
@Header('Content-Type', HLS_PLAYLIST_CONTENT_TYPE)
@ApiProduces(HLS_PLAYLIST_CONTENT_TYPE)
@Endpoint({
summary: 'Get HLS media playlist',
description: 'Returns an HLS media playlist for one variant of the streaming session.',
history: new HistoryBuilder().added('v3').alpha('v3'),
})
getMediaPlaylist(@Auth() auth: AuthDto, @Param() { id, sessionId }: HlsVariantParamDto) {
return this.service.getMediaPlaylist(auth, id, sessionId);
}
@Get(':id/video/stream/:sessionId/:variantIndex/:filename')
@FileResponse()
@Authenticated({ permission: Permission.AssetView, sharedLink: true })
@Endpoint({
summary: 'Get HLS segment or init file',
description: 'Streams an HLS init segment (init.mp4) or media segment (seg_N.m4s).',
history: new HistoryBuilder().added('v3').alpha('v3'),
})
async getSegment(
@Auth() auth: AuthDto,
@Param() { id, sessionId, variantIndex, filename }: HlsSegmentParamDto,
@Res() res: Response,
@Next() next: NextFunction,
) {
await sendFile(res, next, () => this.service.getSegment(auth, id, sessionId, variantIndex, filename), this.logger);
}
@Delete(':id/video/stream/:sessionId')
@HttpCode(HttpStatus.NO_CONTENT)
@Authenticated({ permission: Permission.AssetView, sharedLink: true })
@Endpoint({
summary: 'End HLS streaming session',
description: 'Releases server resources for the streaming session.',
history: new HistoryBuilder().added('v3').alpha('v3'),
})
async endSession(@Auth() auth: AuthDto, @Param() { id, sessionId }: HlsSessionParamDto) {
await this.service.endSession(auth, id, sessionId);
}
}
@@ -1,5 +1,5 @@
import { WorkflowTrigger } from '@immich/plugin-sdk';
import { WorkflowController } from 'src/controllers/workflow.controller';
import { WorkflowTrigger } from 'src/enum';
import { LoggingRepository } from 'src/repositories/logging.repository';
import { WorkflowService } from 'src/services/workflow.service';
import request from 'supertest';
+12
View File
@@ -35,6 +35,10 @@ export interface MoveRequest {
export type ThumbnailPathEntity = { id: string; ownerId: string };
export type HlsSessionFolder = { ownerId: string; sessionId: string };
export type HlsVariantFolder = { ownerId: string; sessionId: string; variantIndex: number };
export type ImagePathOptions = { fileType: AssetFileType; format: ImageFormat | RawExtractedFormat; isEdited: boolean };
let instance: StorageCore | null;
@@ -125,6 +129,14 @@ export class StorageCore {
return StorageCore.getNestedPath(StorageFolder.EncodedVideo, asset.ownerId, `${asset.id}.mp4`);
}
static getHlsSessionFolder({ ownerId, sessionId }: HlsSessionFolder) {
return StorageCore.getNestedPath(StorageFolder.EncodedVideo, ownerId, sessionId);
}
static getHlsVariantFolder({ ownerId, sessionId, variantIndex }: HlsVariantFolder) {
return join(StorageCore.getHlsSessionFolder({ ownerId, sessionId }), variantIndex.toString());
}
static getAndroidMotionPath(asset: ThumbnailPathEntity, uuid: string) {
return StorageCore.getNestedPath(StorageFolder.EncodedVideo, asset.ownerId, `${uuid}-MP.mp4`);
}
+10
View File
@@ -265,3 +265,13 @@ export class HistoryBuilder {
return this;
}
}
// eslint-disable-next-line @typescript-eslint/no-unsafe-function-type
export const extraModels: Function[] = [];
export const ExtraModel = (): ClassDecorator => {
// eslint-disable-next-line unicorn/consistent-function-scoping, @typescript-eslint/no-unsafe-function-type
return (object: Function) => {
extraModels.push(object);
};
};
+2
View File
@@ -65,6 +65,8 @@ const UpdateAlbumSchema = z
const GetAlbumsSchema = z
.object({
id: z.uuidv4().optional().describe('Album ID'),
name: z.string().optional().describe('Album name (exact match)'),
isOwned: stringToBool
.optional()
.describe('Filter by ownership: true = only owned, false = only shared-with-me, undefined = no filter'),
+1
View File
@@ -38,6 +38,7 @@ const PluginManifestTemplateSchema = z
description: z.string().min(1).describe('Template description'),
trigger: WorkflowTriggerSchema.describe('Workflow trigger'),
steps: z.array(PluginManifestTemplateStepSchema).describe('Workflow steps'),
uiHints: z.array(z.string()).optional().default([]).describe('Ui hints, for example "smart-album"'),
})
.meta({ id: 'PluginManifestTemplateDto' });
+5 -1
View File
@@ -1,6 +1,7 @@
import { WorkflowTrigger } from '@immich/plugin-sdk';
import { createZodDto } from 'nestjs-zod';
import { JsonSchemaDto } from 'src/dtos/json-schema.dto';
import { WorkflowTrigger, WorkflowTriggerSchema, WorkflowType, WorkflowTypeSchema } from 'src/enum';
import { WorkflowTriggerSchema, WorkflowType, WorkflowTypeSchema } from 'src/enum';
import { asPluginKey } from 'src/utils/workflow';
import z from 'zod';
@@ -58,6 +59,7 @@ const PluginTemplateResponseSchema = z
description: z.string().describe('Template description'),
trigger: WorkflowTriggerSchema.describe('Workflow trigger'),
steps: z.array(PluginTemplateStepResponseSchema).describe('Workflow steps'),
uiHints: z.array(z.string()).describe('Ui hints, for example "smart-album"'),
})
.meta({ id: 'PluginTemplateResponseDto' });
@@ -91,6 +93,7 @@ export type PluginTemplate = {
config?: Record<string, unknown> | null;
enabled?: boolean;
}>;
uiHints: string[];
};
export const mapTemplate = (plugin: { name: string }, template: PluginTemplate): PluginTemplateResponseDto => {
@@ -104,6 +107,7 @@ export const mapTemplate = (plugin: { name: string }, template: PluginTemplate):
config: step.config ?? null,
enabled: step.enabled,
})),
uiHints: template.uiHints ?? [],
};
};
+40 -11
View File
@@ -1,5 +1,6 @@
import { createZodDto } from 'nestjs-zod';
import type { SemVer } from 'semver';
import { ExtraModel, HistoryBuilder } from 'src/decorators';
import { isoDatetimeToDate } from 'src/validation';
import z from 'zod';
@@ -58,9 +59,15 @@ const ServerStorageResponseSchema = z
const ServerVersionResponseSchema = z
.object({
major: z.int().describe('Major version number'),
minor: z.int().describe('Minor version number'),
patch: z.int().describe('Patch version number'),
major: z.int().min(0).describe('Major version number'),
minor: z.int().min(0).describe('Minor version number'),
patch: z.int().min(0).describe('Patch version number'),
prerelease: z
.int()
.min(0)
.nullable()
.meta(HistoryBuilder.v3().getExtensions())
.describe('Pre-release version number'),
})
.meta({ id: 'ServerVersionResponseDto' });
@@ -117,6 +124,7 @@ const ServerConfigSchema = z
mapDarkStyleUrl: z.string().describe('Map dark style URL'),
mapLightStyleUrl: z.string().describe('Map light style URL'),
maintenanceMode: z.boolean().describe('Whether maintenance mode is active'),
minFaces: z.int().describe('People min faces server default'),
})
.meta({ id: 'ServerConfigDto' });
@@ -137,9 +145,30 @@ const ServerFeaturesSchema = z
search: z.boolean().describe('Whether search is enabled'),
email: z.boolean().describe('Whether email notifications are enabled'),
ocr: z.boolean().describe('Whether OCR is enabled'),
realtimeTranscoding: z.boolean().describe('Whether real-time transcoding is enabled'),
})
.meta({ id: 'ServerFeaturesDto' });
export enum ReleaseType {
Major = 'major',
Premajor = 'premajor',
Minor = 'minor',
Preminor = 'preminor',
Patch = 'patch',
Prepatch = 'prepatch',
Prerelease = 'prerelease',
}
const ReleaseTypeSchema = z.enum(ReleaseType).meta({ id: 'ReleaseType' }).describe('Release type');
const ReleaseEventV1Schema = z.object({
isAvailable: z.boolean().describe('Whether a new version is available'),
checkedAt: z.string().describe('When the server last checked for a latest version. As an ISO timestamp'),
serverVersion: ServerVersionResponseSchema,
releaseVersion: ServerVersionResponseSchema,
type: ReleaseTypeSchema.nullable(),
});
export class ServerPingResponse extends createZodDto(ServerPingResponseSchema) {}
export class ServerAboutResponseDto extends createZodDto(ServerAboutResponseSchema) {}
export class ServerApkLinksDto extends createZodDto(ServerApkLinksSchema) {}
@@ -147,7 +176,12 @@ export class ServerStorageResponseDto extends createZodDto(ServerStorageResponse
export class ServerVersionResponseDto extends createZodDto(ServerVersionResponseSchema) {
static fromSemVer(value: SemVer): z.infer<typeof ServerVersionResponseSchema> {
return { major: value.major, minor: value.minor, patch: value.patch };
return {
major: value.major,
minor: value.minor,
patch: value.patch,
prerelease: (value.prerelease[1] as number) ?? null,
};
}
}
@@ -158,10 +192,5 @@ export class ServerMediaTypesResponseDto extends createZodDto(ServerMediaTypesRe
export class ServerConfigDto extends createZodDto(ServerConfigSchema) {}
export class ServerFeaturesDto extends createZodDto(ServerFeaturesSchema) {}
export interface ReleaseNotification {
isAvailable: boolean;
/** ISO8601 */
checkedAt: string;
serverVersion: ServerVersionResponseDto;
releaseVersion: ServerVersionResponseDto;
}
@ExtraModel()
export class ReleaseEventV1 extends createZodDto(ReleaseEventV1Schema) {}
+26
View File
@@ -0,0 +1,26 @@
import { createZodDto } from 'nestjs-zod';
import z from 'zod';
const HlsSessionParamSchema = z.object({
id: z.uuidv4(),
sessionId: z.uuidv4(),
});
export class HlsSessionParamDto extends createZodDto(HlsSessionParamSchema) {}
const HlsVariantParamSchema = z.object({
id: z.uuidv4(),
sessionId: z.uuidv4(),
variantIndex: z.coerce.number().int().min(0),
});
export class HlsVariantParamDto extends createZodDto(HlsVariantParamSchema) {}
const HlsSegmentParamSchema = z.object({
id: z.uuidv4(),
sessionId: z.uuidv4(),
variantIndex: z.coerce.number().int().min(0),
filename: z.string().regex(/^(init\.mp4|seg_\d+\.m4s)$/, { error: 'Invalid HLS segment filename' }),
});
export class HlsSegmentParamDto extends createZodDto(HlsSegmentParamSchema) {}
+1 -10
View File
@@ -1,5 +1,5 @@
/* eslint-disable @typescript-eslint/no-unsafe-function-type */
import { createZodDto } from 'nestjs-zod';
import { ExtraModel } from 'src/decorators';
import { AssetEditActionSchema } from 'src/dtos/editing.dto';
import {
AlbumUserRole,
@@ -17,15 +17,6 @@ import {
import { isoDatetimeToDate } from 'src/validation';
import z from 'zod';
export const extraSyncModels: Function[] = [];
const ExtraModel = (): ClassDecorator => {
// eslint-disable-next-line unicorn/consistent-function-scoping
return (object: Function) => {
extraSyncModels.push(object);
};
};
const SyncUserV1Schema = z
.object({
id: z.string().describe('User ID'),
+13 -1
View File
@@ -79,6 +79,11 @@ const SystemConfigFFmpegSchema = z
accel: TranscodeHardwareAccelerationSchema,
accelDecode: configBool.describe('Accelerated decode'),
tonemap: ToneMappingSchema,
realtime: z
.object({
enabled: configBool.describe('Enable real-time HLS transcoding (alpha)'),
})
.meta({ id: 'SystemConfigFFmpegRealtimeDto' }),
})
.meta({ id: 'SystemConfigFFmpegDto' });
@@ -151,8 +156,15 @@ const SystemConfigMapSchema = z
})
.meta({ id: 'SystemConfigMapDto' });
export enum ReleaseChannel {
Stable = 'stable',
ReleaseCandidate = 'releaseCandidate',
}
const ReleaseChannelSchema = z.enum(ReleaseChannel).describe('Release channel').meta({ id: 'ReleaseChannel' });
const SystemConfigNewVersionCheckSchema = z
.object({ enabled: configBool.describe('Enabled') })
.object({ enabled: configBool.describe('Enabled'), channel: ReleaseChannelSchema })
.meta({ id: 'SystemConfigNewVersionCheckDto' });
const SystemConfigNightlyTasksSchema = z
+2
View File
@@ -45,6 +45,7 @@ const PeopleUpdateSchema = z
.object({
enabled: z.boolean().optional().describe('Whether people are enabled'),
sidebarWeb: z.boolean().optional().describe('Whether people appear in web sidebar'),
minimumFaces: z.int().min(1).optional().describe('People face threshold'),
})
.optional()
.meta({ id: 'PeopleUpdate' });
@@ -138,6 +139,7 @@ const PeopleResponseSchema = z
.object({
enabled: z.boolean().describe('Whether people are enabled'),
sidebarWeb: z.boolean().describe('Whether people appear in web sidebar'),
minimumFaces: z.int().min(1).optional().describe('People face threshold'),
})
.meta({ id: 'PeopleResponse' });
+2 -2
View File
@@ -1,6 +1,6 @@
import type { WorkflowStepConfig } from '@immich/plugin-sdk';
import type { WorkflowStepConfig, WorkflowTrigger } from '@immich/plugin-sdk';
import { createZodDto } from 'nestjs-zod';
import { WorkflowTrigger, WorkflowTriggerSchema, WorkflowTypeSchema } from 'src/enum';
import { WorkflowTriggerSchema, WorkflowTypeSchema } from 'src/enum';
import z from 'zod';
const WorkflowTriggerResponseSchema = z
+6 -11
View File
@@ -1,3 +1,4 @@
import { WorkflowTrigger } from '@immich/plugin-sdk';
import z from 'zod';
export enum AuthType {
@@ -452,11 +453,7 @@ export enum VideoCodec {
export const VideoCodecSchema = z.enum(VideoCodec).describe('Target video codec').meta({ id: 'VideoCodec' });
export enum VideoSegmentCodec {
Av1 = 'av1',
Hevc = 'hevc',
H264 = 'h264',
}
export type VideoSegmentCodec = VideoCodec.Av1 | VideoCodec.Hevc | VideoCodec.H264;
export enum AudioCodec {
Mp3 = 'mp3',
@@ -826,6 +823,8 @@ export enum JobName {
LibrarySyncFiles = 'LibrarySyncFiles',
LibraryScanQueueAll = 'LibraryScanQueueAll',
HlsSessionCleanup = 'HlsSessionCleanup',
MemoryCleanup = 'MemoryCleanup',
MemoryGenerate = 'MemoryGenerate',
@@ -866,7 +865,7 @@ export enum JobName {
Ocr = 'Ocr',
// Workflow
WorkflowAssetCreate = 'WorkflowAssetCreate',
WorkflowAssetTrigger = 'WorkflowAssetTrigger',
}
export const JobNameSchema = z.enum(JobName).describe('Job name').meta({ id: 'JobName' });
@@ -919,6 +918,7 @@ export enum DatabaseLock {
MaintenanceOperation = 621,
MemoryCreation = 777,
VersionCheck = 800,
HlsSessionCleanup = 850,
}
export enum MaintenanceAction {
@@ -1164,11 +1164,6 @@ export enum PluginContext {
export const PluginContextSchema = z.enum(PluginContext).describe('Plugin context').meta({ id: 'PluginContextType' });
export enum WorkflowTrigger {
AssetCreate = 'AssetCreate',
PersonRecognized = 'PersonRecognized',
}
export const WorkflowTriggerSchema = z
.enum(WorkflowTrigger)
.describe('Plugin trigger type')
+12 -1
View File
@@ -42,7 +42,18 @@ group by
having
(
"person"."name" != $3
or count("asset_face"."assetId") >= $4
or count("asset_face"."assetId") >= COALESCE(
(
SELECT
value -> 'people' ->> 'minimumFaces'
FROM
user_metadata
WHERE
"userId" = $4
AND key = 'preferences'
),
'3'
)::int
)
order by
"person"."isHidden" asc,
+255 -2
View File
@@ -7,6 +7,7 @@ from
"video_stream_session"
where
"id" = $1
and "expiresAt" > $2
-- VideoStreamRepository.getVariant
select
@@ -27,11 +28,13 @@ where
-- VideoStreamRepository.getExpiredSessions
select
"id"
"video_stream_session"."id",
"asset"."ownerId"
from
"video_stream_session"
inner join "asset" on "asset"."id" = "video_stream_session"."assetId"
where
"expiresAt" <= $1
"video_stream_session"."expiresAt" <= $1
-- VideoStreamRepository.extendSession
update "video_stream_session"
@@ -44,3 +47,253 @@ where
delete from "video_stream_session"
where
"id" = $1
-- VideoStreamRepository.getForMainPlaylist
select
(
select
to_json(obj)
from
(
select
"asset_video"."index",
"asset_video"."codecName",
"asset_video"."profile",
"asset_video"."level",
"asset_video"."bitrate",
"asset_exif"."exifImageWidth" as "width",
"asset_exif"."exifImageHeight" as "height",
"asset_video"."pixelFormat",
"asset_video"."frameCount",
"asset_exif"."fps" as "frameRate",
"asset_video"."timeBase",
case
when "asset_exif"."orientation" = '6' then -90
when "asset_exif"."orientation" = '8' then 90
when "asset_exif"."orientation" = '3' then 180
else 0
end as "rotation",
"asset_video"."colorPrimaries",
"asset_video"."colorMatrix",
"asset_video"."colorTransfer",
"asset_video"."dvProfile",
"asset_video"."dvLevel",
"asset_video"."dvBlSignalCompatibilityId"
from
(
select
1
) as "dummy"
where
"asset_video"."assetId" is not null
) as obj
) as "videoStream",
(
select
to_json(obj)
from
(
select
"asset_keyframe"."pts" as "keyframePts",
"asset_keyframe"."accDuration" as "keyframeAccDuration",
"asset_keyframe"."ownDuration" as "keyframeOwnDuration",
"asset_keyframe"."totalDuration",
"asset_keyframe"."packetCount",
"asset_keyframe"."outputFrames"
from
(
select
1
) as "dummy"
where
"asset_keyframe"."assetId" is not null
) as obj
) as "packets"
from
"asset"
inner join "asset_exif" on "asset"."id" = "asset_exif"."assetId"
inner join "asset_video" on "asset"."id" = "asset_video"."assetId"
inner join "asset_keyframe" on "asset"."id" = "asset_keyframe"."assetId"
where
"asset"."id" = $1
-- VideoStreamRepository.getForMediaPlaylist
select
(
select
to_json(obj)
from
(
select
"asset_video"."index",
"asset_video"."codecName",
"asset_video"."profile",
"asset_video"."level",
"asset_video"."bitrate",
"asset_exif"."exifImageWidth" as "width",
"asset_exif"."exifImageHeight" as "height",
"asset_video"."pixelFormat",
"asset_video"."frameCount",
"asset_exif"."fps" as "frameRate",
"asset_video"."timeBase",
case
when "asset_exif"."orientation" = '6' then -90
when "asset_exif"."orientation" = '8' then 90
when "asset_exif"."orientation" = '3' then 180
else 0
end as "rotation",
"asset_video"."colorPrimaries",
"asset_video"."colorMatrix",
"asset_video"."colorTransfer",
"asset_video"."dvProfile",
"asset_video"."dvLevel",
"asset_video"."dvBlSignalCompatibilityId"
from
(
select
1
) as "dummy"
where
"asset_video"."assetId" is not null
) as obj
) as "videoStream",
(
select
to_json(obj)
from
(
select
"asset_keyframe"."pts" as "keyframePts",
"asset_keyframe"."accDuration" as "keyframeAccDuration",
"asset_keyframe"."ownDuration" as "keyframeOwnDuration",
"asset_keyframe"."totalDuration",
"asset_keyframe"."packetCount",
"asset_keyframe"."outputFrames"
from
(
select
1
) as "dummy"
where
"asset_keyframe"."assetId" is not null
) as obj
) as "packets"
from
"asset"
inner join "asset_exif" on "asset"."id" = "asset_exif"."assetId"
inner join "video_stream_session" on "asset"."id" = "video_stream_session"."assetId"
inner join "asset_video" on "asset"."id" = "asset_video"."assetId"
inner join "asset_keyframe" on "asset"."id" = "asset_keyframe"."assetId"
where
"asset"."id" = $1
and "video_stream_session"."id" = $2
and "video_stream_session"."expiresAt" > $3
-- VideoStreamRepository.getForTranscoding
select
"asset"."originalPath",
(
select
to_json(obj)
from
(
select
"asset_audio"."index",
"asset_audio"."codecName",
"asset_audio"."profile",
"asset_audio"."bitrate"
from
(
select
1
) as "dummy"
where
"asset_audio"."assetId" is not null
) as obj
) as "audioStream",
(
select
to_json(obj)
from
(
select
"asset_video"."index",
"asset_video"."codecName",
"asset_video"."profile",
"asset_video"."level",
"asset_video"."bitrate",
"asset_exif"."exifImageWidth" as "width",
"asset_exif"."exifImageHeight" as "height",
"asset_video"."pixelFormat",
"asset_video"."frameCount",
"asset_exif"."fps" as "frameRate",
"asset_video"."timeBase",
case
when "asset_exif"."orientation" = '6' then -90
when "asset_exif"."orientation" = '8' then 90
when "asset_exif"."orientation" = '3' then 180
else 0
end as "rotation",
"asset_video"."colorPrimaries",
"asset_video"."colorMatrix",
"asset_video"."colorTransfer",
"asset_video"."dvProfile",
"asset_video"."dvLevel",
"asset_video"."dvBlSignalCompatibilityId"
from
(
select
1
) as "dummy"
where
"asset_video"."assetId" is not null
) as obj
) as "videoStream",
(
select
to_json(obj)
from
(
select
"asset_video"."formatName",
"asset_video"."formatLongName",
"asset"."duration",
"asset_video"."bitrate"
from
(
select
1
) as "dummy"
where
"asset_video"."assetId" is not null
) as obj
) as "format",
(
select
to_json(obj)
from
(
select
"asset_keyframe"."pts" as "keyframePts",
"asset_keyframe"."accDuration" as "keyframeAccDuration",
"asset_keyframe"."ownDuration" as "keyframeOwnDuration",
"asset_keyframe"."totalDuration",
"asset_keyframe"."packetCount",
"asset_keyframe"."outputFrames"
from
(
select
1
) as "dummy"
where
"asset_keyframe"."assetId" is not null
) as obj
) as "packets"
from
"asset"
inner join "asset_exif" on "asset"."id" = "asset_exif"."assetId"
left join "asset_audio" on "asset"."id" = "asset_audio"."assetId"
inner join "asset_video" on "asset"."id" = "asset_video"."assetId"
inner join "asset_keyframe" on "asset"."id" = "asset_keyframe"."assetId"
where
"asset"."id" = $1
+6 -1
View File
@@ -209,11 +209,16 @@ export class AlbumRepository {
}
@GenerateSql({ params: [DummyValue.UUID, { isOwned: true, isShared: true }] })
getAll(ownerId: string, options: { isOwned?: boolean; isShared?: boolean } = {}): Promise<MapAlbumDto[]> {
getAll(
ownerId: string,
options: { id?: string; isOwned?: boolean; isShared?: boolean; name?: string } = {},
): Promise<MapAlbumDto[]> {
return this.buildAlbumBaseQuery(ownerId, options)
.selectAll('album')
.select(withAlbumUsers(ownerId))
.select(withSharedLink)
.$if(!!options.id, (qb) => qb.where('album.id', '=', options.id!))
.$if(!!options.name, (qb) => qb.where('album.albumName', '=', options.name!))
.orderBy('album.createdAt', 'desc')
.execute();
}
@@ -92,6 +92,14 @@ type EventMap = {
AuthChangePassword: [{ userId: string; currentSessionId?: string; invalidateSessions?: boolean }];
// hls streaming events
HlsSegmentRequest: [{ sessionId: string; assetId: string; variantIndex: number; segmentIndex: number }];
HlsSegmentResult: [{ sessionId: string; variantIndex: number; segmentIndex: number; error?: string }];
HlsHeartbeat: [{ sessionId: string; variantIndex?: number; segmentIndex?: number }];
HlsSessionRequest: [{ sessionId: string; assetId: string; ownerId: string }];
HlsSessionResult: [{ sessionId: string; error?: string }];
HlsSessionEnd: [{ sessionId: string }];
// websocket events
WebsocketConnect: [{ userId: string }];
};
+29 -4
View File
@@ -490,18 +490,43 @@ export class MediaRepository {
return this.parseInt(b.bit_rate) - this.parseInt(a.bit_rate);
}
/* Ported from https://code.ffmpeg.org/FFmpeg/FFmpeg/src/commit/5c44245878e235ae64fe87fb9877644856d33d1d/fftools/ffmpeg_filter.c
* SPDX-License-Identifier: LGPL-2.1-or-later
* Copyright (c) FFmpeg authors and contributors — https://ffmpeg.org/
* Modifications: TS port operating on probe-derived packet metadata rather than decoded AVFrames. */
private cfrOutputFrames(packets: { pts: number; duration: number }[], slotsPerTick: number) {
// Packets may be out of PTS order due to B-frames
packets.sort((a, b) => a.pts - b.pts);
const firstPts = packets[0].pts;
let outputFrames = 0;
let nextPts = 0;
const history = [0, 0, 0];
for (const pkt of packets) {
const delta = (pkt.pts - firstPts) * slotsPerTick - nextPts + pkt.duration * slotsPerTick;
const nb = delta < -1.1 ? 0 : delta > 1.1 ? Math.round(delta) : 1;
const syncIpts = (pkt.pts - firstPts) * slotsPerTick;
const duration = pkt.duration * slotsPerTick;
let delta0 = syncIpts - nextPts;
const delta = delta0 + duration;
if (delta0 < 0 && delta > 0) {
delta0 = 0;
}
let nb = 1;
let nbPrev = 0;
if (delta < -1.1) {
nb = 0;
} else if (delta > 1.1) {
nb = Math.round(delta);
if (delta0 > 1.1) {
nbPrev = Math.round(delta0 - 0.6);
}
}
outputFrames += nb;
nextPts += nb;
history[2] = history[1];
history[1] = history[0];
history[0] = nbPrev;
}
return outputFrames;
const median = history.sort((a, b) => a - b)[1];
return outputFrames + median;
}
}
+12 -3
View File
@@ -4,7 +4,7 @@ import { jsonObjectFrom } from 'kysely/helpers/postgres';
import { InjectKysely } from 'nestjs-kysely';
import { AssetFace } from 'src/database';
import { Chunked, ChunkedArray, DummyValue, GenerateSql } from 'src/decorators';
import { AssetFileType, AssetVisibility, SourceType } from 'src/enum';
import { AssetFileType, AssetVisibility, SourceType, UserMetadataKey } from 'src/enum';
import { DB } from 'src/schema';
import { AssetFaceTable } from 'src/schema/tables/asset-face.table';
import { FaceSearchTable } from 'src/schema/tables/face-search.table';
@@ -13,7 +13,6 @@ import { dummy, removeUndefinedKeys, withFilePath } from 'src/utils/database';
import { paginationHelper, PaginationOptions } from 'src/utils/pagination';
export interface PersonSearchOptions {
minimumFaceCount: number;
withHidden: boolean;
closestFaceAssetId?: string;
}
@@ -168,7 +167,17 @@ export class PersonRepository {
.having((eb) =>
eb.or([
eb('person.name', '!=', ''),
eb((innerEb) => innerEb.fn.count('asset_face.assetId'), '>=', options?.minimumFaceCount || 1),
eb(
(innerEb) => innerEb.fn.count('asset_face.assetId'),
'>=',
sql<number>`COALESCE(
(SELECT value -> 'people' ->> 'minimumFaces'
FROM user_metadata
WHERE "userId" = ${userId}
AND key = ${sql.lit(UserMetadataKey.Preferences)}),
'3'
)::int `,
),
]),
)
.groupBy('person.id')
@@ -1,12 +1,10 @@
import { Injectable } from '@nestjs/common';
import { ChildProcessWithoutNullStreams, fork, spawn, SpawnOptionsWithoutStdio } from 'node:child_process';
import { fork, spawn, SpawnOptionsWithoutStdio } from 'node:child_process';
import { Duplex } from 'node:stream';
@Injectable()
export class ProcessRepository {
spawn(command: string, args?: readonly string[], options?: SpawnOptionsWithoutStdio): ChildProcessWithoutNullStreams {
return spawn(command, args, options);
}
spawn = spawn;
spawnDuplexStream(command: string, args?: readonly string[], options?: SpawnOptionsWithoutStdio): Duplex {
let stdinClosed = false;
@@ -4,6 +4,7 @@ import { exec as execCallback } from 'node:child_process';
import { readFile } from 'node:fs/promises';
import { promisify } from 'node:util';
import sharp from 'sharp';
import { ReleaseChannel } from 'src/dtos/system-config.dto';
import { ConfigRepository } from 'src/repositories/config.repository';
import { LoggingRepository } from 'src/repositories/logging.repository';
@@ -64,10 +65,12 @@ export class ServerInfoRepository {
this.logger.setContext(ServerInfoRepository.name);
}
async getLatestRelease(): Promise<VersionResponse> {
async getLatestRelease(channel: ReleaseChannel): Promise<VersionResponse> {
try {
const { versionCheck } = this.configRepository.getEnv();
const response = await fetch(versionCheck.url);
const url = new URL(versionCheck.url);
url.searchParams.append('channel', channel);
const response = await fetch(url);
if (!response.ok) {
throw new Error(`Version check request failed with status ${response.status}: ${await response.text()}`);
@@ -10,6 +10,7 @@ import {
existsSync,
mkdirSync,
ReadOptionsWithBuffer,
watch,
} from 'node:fs';
import fs from 'node:fs/promises';
import path from 'node:path';
@@ -277,6 +278,8 @@ export class StorageRepository {
return () => watcher.close();
}
watchDir = watch; // Native fs.watch without chokidar overhead
private asGlob(pathToCrawl: string): string {
const escapedPath = escapePath(pathToCrawl).replaceAll('"', '["]').replaceAll("'", "[']").replaceAll('`', '[`]');
const extensions = `*{${mimeTypes.getSupportedFileExtensions().join(',')}}`;
@@ -8,6 +8,7 @@ import {
VideoStreamSessionTable,
VideoStreamVariantTable,
} from 'src/schema/tables/video-stream.table';
import { withAudioStream, withVideoFormat, withVideoPackets, withVideoStream } from 'src/utils/database';
@Injectable()
export class VideoStreamRepository {
@@ -27,7 +28,12 @@ export class VideoStreamRepository {
@GenerateSql({ params: [DummyValue.UUID] })
getSession(id: string) {
return this.db.selectFrom('video_stream_session').selectAll().where('id', '=', id).executeTakeFirst();
return this.db
.selectFrom('video_stream_session')
.selectAll()
.where('id', '=', id)
.where('expiresAt', '>', new Date())
.executeTakeFirst();
}
@GenerateSql({ params: [DummyValue.UUID] })
@@ -47,7 +53,12 @@ export class VideoStreamRepository {
@GenerateSql()
getExpiredSessions() {
return this.db.selectFrom('video_stream_session').select(['id']).where('expiresAt', '<=', new Date()).execute();
return this.db
.selectFrom('video_stream_session')
.innerJoin('asset', 'asset.id', 'video_stream_session.assetId')
.select(['video_stream_session.id', 'asset.ownerId'])
.where('video_stream_session.expiresAt', '<=', new Date())
.execute();
}
@GenerateSql({ params: [DummyValue.UUID, DummyValue.DATE] })
@@ -59,4 +70,50 @@ export class VideoStreamRepository {
async deleteSession(id: string) {
await this.db.deleteFrom('video_stream_session').where('id', '=', id).execute();
}
@GenerateSql({ params: [DummyValue.UUID] })
async getForMainPlaylist(id: string) {
return this.db
.selectFrom('asset')
.innerJoin('asset_exif', 'asset.id', 'asset_exif.assetId')
.where('asset.id', '=', id)
.innerJoin('asset_video', 'asset.id', 'asset_video.assetId')
.innerJoin('asset_keyframe', 'asset.id', 'asset_keyframe.assetId')
.select((eb) => withVideoStream(eb).$notNull().as('videoStream'))
.select((eb) => withVideoPackets(eb).$notNull().as('packets'))
.executeTakeFirst();
}
@GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID] })
async getForMediaPlaylist(id: string, sessionId: string) {
return this.db
.selectFrom('asset')
.innerJoin('asset_exif', 'asset.id', 'asset_exif.assetId')
.innerJoin('video_stream_session', 'asset.id', 'video_stream_session.assetId')
.where('asset.id', '=', id)
.where('video_stream_session.id', '=', sessionId)
.where('video_stream_session.expiresAt', '>', new Date())
.innerJoin('asset_video', 'asset.id', 'asset_video.assetId')
.innerJoin('asset_keyframe', 'asset.id', 'asset_keyframe.assetId')
.select((eb) => withVideoStream(eb).$notNull().as('videoStream'))
.select((eb) => withVideoPackets(eb).$notNull().as('packets'))
.executeTakeFirst();
}
@GenerateSql({ params: [DummyValue.UUID] })
async getForTranscoding(id: string) {
return this.db
.selectFrom('asset')
.innerJoin('asset_exif', 'asset.id', 'asset_exif.assetId')
.where('asset.id', '=', id)
.leftJoin('asset_audio', 'asset.id', 'asset_audio.assetId')
.innerJoin('asset_video', 'asset.id', 'asset_video.assetId')
.innerJoin('asset_keyframe', 'asset.id', 'asset_keyframe.assetId')
.select('asset.originalPath')
.select((eb) => withAudioStream(eb).as('audioStream'))
.select((eb) => withVideoStream(eb).$notNull().as('videoStream'))
.select((eb) => withVideoFormat(eb).$notNull().as('format'))
.select((eb) => withVideoPackets(eb).$notNull().as('packets'))
.executeTakeFirst();
}
}
@@ -10,13 +10,22 @@ import { Server, Socket } from 'socket.io';
import { AssetResponseDto } from 'src/dtos/asset-response.dto';
import { AuthDto } from 'src/dtos/auth.dto';
import { NotificationDto } from 'src/dtos/notification.dto';
import { ReleaseNotification, ServerVersionResponseDto } from 'src/dtos/server.dto';
import { ReleaseEventV1, ServerVersionResponseDto } from 'src/dtos/server.dto';
import { SyncAssetEditV1, SyncAssetExifV1, SyncAssetV2 } from 'src/dtos/sync.dto';
import { AppRestartEvent, ArgsOf, EventRepository } from 'src/repositories/event.repository';
import { LoggingRepository } from 'src/repositories/logging.repository';
import { handlePromiseError } from 'src/utils/misc';
export const serverEvents = ['ConfigUpdate', 'AppRestart'] as const;
export const serverEvents = [
'ConfigUpdate',
'AppRestart',
'HlsSegmentRequest',
'HlsSegmentResult',
'HlsHeartbeat',
'HlsSessionRequest',
'HlsSessionResult',
'HlsSessionEnd',
] as const;
export type ServerEvents = (typeof serverEvents)[number];
export interface ClientEventMap {
@@ -31,7 +40,7 @@ export interface ClientEventMap {
on_person_thumbnail: [string];
on_server_version: [ServerVersionResponseDto];
on_config_update: [];
on_new_release: [ReleaseNotification];
on_new_release: [ReleaseEventV1];
on_notification: [NotificationDto];
on_session_delete: [string];
@@ -45,10 +45,10 @@ export class WorkflowRepository {
}
@GenerateSql({ params: [DummyValue.UUID] })
search(dto: WorkflowSearchDto & { ownerId?: string }) {
search(dto: WorkflowSearchDto & { userId?: string }) {
return this.queryBuilder()
.$if(!!dto.id, (qb) => qb.where('id', '=', dto.id!))
.$if(!!dto.ownerId, (qb) => qb.where('ownerId', '=', dto.ownerId!))
.$if(!!dto.userId, (qb) => qb.where('ownerId', '=', dto.userId!))
.$if(!!dto.trigger, (qb) => qb.where('trigger', '=', dto.trigger!))
.$if(dto.enabled !== undefined, (qb) => qb.where('enabled', '=', dto.enabled!))
.orderBy('createdAt', 'desc')
@@ -103,6 +103,10 @@ export class WorkflowRepository {
});
}
async updateStep(id: string, dto: Updateable<WorkflowStepTable>) {
await this.db.updateTable('workflow_step').where('workflow_step.id', '=', id).set(dto).execute();
}
private async replaceAndReturn(tx: Kysely<DB>, workflowId: string, steps?: WorkflowStepUpsert[]) {
if (steps) {
await tx.deleteFrom('workflow_step').where('workflowId', '=', workflowId).execute();
+2 -9
View File
@@ -1,12 +1,5 @@
import { registerEnum } from '@immich/sql-tools';
import {
AlbumUserRole,
AssetStatus,
AssetVisibility,
ChecksumAlgorithm,
SourceType,
VideoSegmentCodec,
} from 'src/enum';
import { AlbumUserRole, AssetStatus, AssetVisibility, ChecksumAlgorithm, SourceType, VideoCodec } from 'src/enum';
export const album_user_role_enum = registerEnum({
name: 'album_user_role_enum',
@@ -35,5 +28,5 @@ export const asset_checksum_algorithm_enum = registerEnum({
export const video_stream_variant_codec_enum = registerEnum({
name: 'video_stream_variant_codec_enum',
values: Object.values(VideoSegmentCodec),
values: [VideoCodec.Av1, VideoCodec.Hevc, VideoCodec.H264],
});
@@ -0,0 +1,16 @@
import { Kysely, sql } from 'kysely';
export async function up(db: Kysely<any>): Promise<void> {
// Delete unauthorized cross-owner asset faces
await sql`
DELETE FROM asset_face
USING person, asset
WHERE asset_face."personId" = person.id
AND asset_face."assetId" = asset.id
AND person."ownerId" != asset."ownerId"
`.execute(db);
}
export async function down(): Promise<void> {
// Not implemented: the deleted rows were unauthorized cross-owner entries
}
+1 -1
View File
@@ -1,3 +1,4 @@
import { WorkflowTrigger } from '@immich/plugin-sdk';
import {
Column,
CreateDateColumn,
@@ -9,7 +10,6 @@ import {
UpdateDateColumn,
} from '@immich/sql-tools';
import { UpdatedAtTrigger, UpdateIdColumn } from 'src/decorators';
import { WorkflowTrigger } from 'src/enum';
import { UserTable } from 'src/schema/tables/user.table';
@Table('workflow')
+2 -5
View File
@@ -37,15 +37,12 @@ export class AlbumService extends BaseService {
};
}
async getAll(
{ user: { id: ownerId } }: AuthDto,
{ assetId, isOwned, isShared }: GetAlbumsDto,
): Promise<AlbumResponseDto[]> {
async getAll({ user: { id: ownerId } }: AuthDto, { assetId, ...rest }: GetAlbumsDto): Promise<AlbumResponseDto[]> {
await this.albumRepository.updateThumbnails();
const albums = assetId
? await this.albumRepository.getByAssetId(ownerId, assetId)
: await this.albumRepository.getAll(ownerId, { isOwned, isShared });
: await this.albumRepository.getAll(ownerId, rest);
if (albums.length === 0) {
return [];
+327
View File
@@ -0,0 +1,327 @@
import { BadRequestException, NotFoundException } from '@nestjs/common';
import { TranscodeHardwareAcceleration } from 'src/enum';
import { HlsService } from 'src/services/hls.service';
import { eiffelTower, train, waterfall } from 'test/fixtures/media.stub';
import { factory } from 'test/small.factory';
import { newTestService, ServiceMocks } from 'test/utils';
// EXTINF values come from FFmpeg's playlist to enforce an exact match
const eiffelExpectedMediaPlaylist = `#EXTM3U
#EXT-X-VERSION:7
#EXT-X-TARGETDURATION:2
#EXT-X-MEDIA-SEQUENCE:0
#EXT-X-PLAYLIST-TYPE:VOD
#EXT-X-MAP:URI="init.mp4"
#EXTINF:2.007222,
seg_0.m4s
#EXTINF:2.007222,
seg_1.m4s
#EXTINF:2.007222,
seg_2.m4s
#EXTINF:2.007222,
seg_3.m4s
#EXTINF:2.007222,
seg_4.m4s
#EXTINF:2.007222,
seg_5.m4s
#EXTINF:2.007222,
seg_6.m4s
#EXTINF:2.007222,
seg_7.m4s
#EXTINF:2.007222,
seg_8.m4s
#EXTINF:2.007222,
seg_9.m4s
#EXTINF:2.007222,
seg_10.m4s
#EXTINF:0.281011,
seg_11.m4s
#EXT-X-ENDLIST
`;
const waterfallExpectedMediaPlaylist = `#EXTM3U
#EXT-X-VERSION:7
#EXT-X-TARGETDURATION:2
#EXT-X-MEDIA-SEQUENCE:0
#EXT-X-PLAYLIST-TYPE:VOD
#EXT-X-MAP:URI="init.mp4"
#EXTINF:2.011405,
seg_0.m4s
#EXTINF:2.011405,
seg_1.m4s
#EXTINF:2.011405,
seg_2.m4s
#EXTINF:2.011405,
seg_3.m4s
#EXTINF:2.011405,
seg_4.m4s
#EXTINF:0.301711,
seg_5.m4s
#EXT-X-ENDLIST
`;
const trainExpectedMediaPlaylist = `#EXTM3U
#EXT-X-VERSION:7
#EXT-X-TARGETDURATION:2
#EXT-X-MEDIA-SEQUENCE:0
#EXT-X-PLAYLIST-TYPE:VOD
#EXT-X-MAP:URI="init.mp4"
#EXTINF:2.000000,
seg_0.m4s
#EXTINF:2.000000,
seg_1.m4s
#EXTINF:2.000000,
seg_2.m4s
#EXTINF:2.000000,
seg_3.m4s
#EXTINF:2.000000,
seg_4.m4s
#EXTINF:2.000000,
seg_5.m4s
#EXTINF:2.000000,
seg_6.m4s
#EXTINF:2.000000,
seg_7.m4s
#EXTINF:2.000000,
seg_8.m4s
#EXTINF:2.000000,
seg_9.m4s
#EXTINF:1.733333,
seg_10.m4s
#EXT-X-ENDLIST
`;
const sessionId = '00000000-0000-0000-0000-000000000000';
const eiffelExpectedMasterDisabled = `#EXTM3U
#EXT-X-VERSION:7
#EXT-X-STREAM-INF:BANDWIDTH=1000000,RESOLUTION=480x852,CODECS="av01.0.04M.08,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/0/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=1200000,RESOLUTION=480x852,CODECS="hvc1.1.6.L90.B0,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/1/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2500000,RESOLUTION=480x852,CODECS="avc1.64001e,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/2/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2000000,RESOLUTION=720x1280,CODECS="av01.0.08M.08,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/3/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2500000,RESOLUTION=720x1280,CODECS="hvc1.1.6.L93.B0,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/4/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=5000000,RESOLUTION=720x1280,CODECS="avc1.64001f,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/5/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=4000000,RESOLUTION=1080x1920,CODECS="av01.0.09M.08,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/6/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=4500000,RESOLUTION=1080x1920,CODECS="hvc1.1.6.L120.B0,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/7/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=8000000,RESOLUTION=1080x1920,CODECS="avc1.640028,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/8/playlist.m3u8
`;
const eiffelExpectedMasterRkmpp = `#EXTM3U
#EXT-X-VERSION:7
#EXT-X-STREAM-INF:BANDWIDTH=1200000,RESOLUTION=480x852,CODECS="hvc1.1.6.L90.B0,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/1/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2500000,RESOLUTION=480x852,CODECS="avc1.64001e,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/2/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2500000,RESOLUTION=720x1280,CODECS="hvc1.1.6.L93.B0,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/4/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=5000000,RESOLUTION=720x1280,CODECS="avc1.64001f,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/5/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=4500000,RESOLUTION=1080x1920,CODECS="hvc1.1.6.L120.B0,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/7/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=8000000,RESOLUTION=1080x1920,CODECS="avc1.640028,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=24.910
${sessionId}/8/playlist.m3u8
`;
const waterfallExpectedMasterDisabled = `#EXTM3U
#EXT-X-VERSION:7
#EXT-X-STREAM-INF:BANDWIDTH=1000000,RESOLUTION=480x852,CODECS="av01.0.04M.08,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=29.830
${sessionId}/0/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=1200000,RESOLUTION=480x852,CODECS="hvc1.1.6.L90.B0,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=29.830
${sessionId}/1/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2500000,RESOLUTION=480x852,CODECS="avc1.64001e,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=29.830
${sessionId}/2/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2000000,RESOLUTION=720x1280,CODECS="av01.0.08M.08,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=29.830
${sessionId}/3/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2500000,RESOLUTION=720x1280,CODECS="hvc1.1.6.L93.B0,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=29.830
${sessionId}/4/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=5000000,RESOLUTION=720x1280,CODECS="avc1.64001f,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=29.830
${sessionId}/5/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=4000000,RESOLUTION=1080x1920,CODECS="av01.0.09M.08,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=29.830
${sessionId}/6/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=4500000,RESOLUTION=1080x1920,CODECS="hvc1.1.6.L120.B0,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=29.830
${sessionId}/7/playlist.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=8000000,RESOLUTION=1080x1920,CODECS="avc1.640028,mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=29.830
${sessionId}/8/playlist.m3u8
`;
describe(HlsService.name, () => {
let sut: HlsService;
let mocks: ServiceMocks;
beforeEach(() => {
({ sut, mocks } = newTestService(HlsService));
});
describe('getMainPlaylist', () => {
const auth = factory.auth();
const assetId = 'asset-1';
const setup = (asset: typeof eiffelTower | typeof waterfall, accel: TranscodeHardwareAcceleration) => {
mocks.access.asset.checkOwnerAccess.mockResolvedValue(new Set([assetId]));
mocks.systemMetadata.get.mockResolvedValue({ ffmpeg: { realtime: { enabled: true }, accel } });
mocks.videoStream.getForMainPlaylist.mockResolvedValue(asset);
mocks.crypto.randomUUID.mockReturnValue(sessionId);
mocks.websocket.serverSend.mockImplementation((event, ...rest) => {
if (event === 'HlsSessionRequest') {
const { sessionId: id } = rest[0] as { sessionId: string };
queueMicrotask(() => sut.onSessionResult({ sessionId: id }));
}
});
};
it('returns main playlist for eiffel-tower (1080p portrait, no acceleration)', async () => {
setup(eiffelTower, TranscodeHardwareAcceleration.Disabled);
await expect(sut.getMainPlaylist(auth, assetId)).resolves.toBe(eiffelExpectedMasterDisabled);
});
it('returns main playlist for eiffel-tower with RKMPP (no AV1 variants)', async () => {
setup(eiffelTower, TranscodeHardwareAcceleration.Rkmpp);
await expect(sut.getMainPlaylist(auth, assetId)).resolves.toBe(eiffelExpectedMasterRkmpp);
});
it('returns main playlist for waterfall (4K landscape) with no acceleration', async () => {
setup(waterfall, TranscodeHardwareAcceleration.Disabled);
await expect(sut.getMainPlaylist(auth, assetId)).resolves.toBe(waterfallExpectedMasterDisabled);
});
it('throws BadRequestException when realtime transcoding is disabled', async () => {
mocks.access.asset.checkOwnerAccess.mockResolvedValue(new Set([assetId]));
mocks.systemMetadata.get.mockResolvedValue({ ffmpeg: { realtime: { enabled: false } } });
await expect(sut.getMainPlaylist(auth, assetId)).rejects.toBeInstanceOf(BadRequestException);
});
it('throws NotFoundException when asset is not yet ready for streaming', async () => {
mocks.access.asset.checkOwnerAccess.mockResolvedValue(new Set([assetId]));
mocks.systemMetadata.get.mockResolvedValue({ ffmpeg: { realtime: { enabled: true } } });
await expect(sut.getMainPlaylist(auth, assetId)).rejects.toBeInstanceOf(NotFoundException);
});
});
describe('getMediaPlaylist', () => {
const auth = factory.auth();
const assetId = 'asset-1';
const fixtures = [
{ data: eiffelTower, playlist: eiffelExpectedMediaPlaylist },
{ data: waterfall, playlist: waterfallExpectedMediaPlaylist },
{ data: train, playlist: trainExpectedMediaPlaylist },
];
it.each(fixtures)('matches FFmpeg for $data.originalPath', async ({ data, playlist }) => {
mocks.access.asset.checkOwnerAccess.mockResolvedValue(new Set([assetId]));
mocks.videoStream.getForMediaPlaylist.mockResolvedValue(data);
await expect(sut.getMediaPlaylist(auth, assetId, sessionId)).resolves.toBe(playlist);
});
it('throws NotFoundException when the session/asset cannot be loaded', async () => {
mocks.access.asset.checkOwnerAccess.mockResolvedValue(new Set([assetId]));
await expect(sut.getMediaPlaylist(auth, assetId, sessionId)).rejects.toBeInstanceOf(NotFoundException);
});
});
describe('getSegment', () => {
const auth = factory.auth();
const assetId = 'asset-1';
const variantIndex = 0;
beforeEach(() => {
mocks.access.asset.checkOwnerAccess.mockResolvedValue(new Set([assetId]));
mocks.videoStream.getSession.mockResolvedValue({ id: sessionId, assetId } as never);
mocks.storage.checkFileExists.mockResolvedValue(true);
});
it('emits HlsHeartbeat with segmentIndex 0 for the first init.mp4 request', async () => {
await sut.getSegment(auth, assetId, sessionId, variantIndex, 'init.mp4');
expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsHeartbeat', {
sessionId,
variantIndex,
segmentIndex: 0,
});
});
it('emits HlsHeartbeat with the parsed segment number for seg_K.m4s', async () => {
await sut.getSegment(auth, assetId, sessionId, variantIndex, 'seg_5.m4s');
expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsHeartbeat', {
sessionId,
variantIndex,
segmentIndex: 5,
});
});
it('returns lastRequested + 1 for init.mp4 after a segment has been served', async () => {
await sut.getSegment(auth, assetId, sessionId, variantIndex, 'seg_5.m4s');
mocks.websocket.serverSend.mockClear();
await sut.getSegment(auth, assetId, sessionId, variantIndex, 'init.mp4');
expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsHeartbeat', {
sessionId,
variantIndex,
segmentIndex: 6,
});
});
it('updates lastRequested on a backward-seek segment request', async () => {
await sut.getSegment(auth, assetId, sessionId, variantIndex, 'seg_5.m4s');
await sut.getSegment(auth, assetId, sessionId, variantIndex, 'seg_3.m4s');
mocks.websocket.serverSend.mockClear();
await sut.getSegment(auth, assetId, sessionId, variantIndex, 'init.mp4');
expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsHeartbeat', {
sessionId,
variantIndex,
segmentIndex: 4,
});
});
it('tracks segment state per session independently', async () => {
await sut.getSegment(auth, assetId, 'session-a', variantIndex, 'seg_5.m4s');
await sut.getSegment(auth, assetId, 'session-b', variantIndex, 'seg_2.m4s');
mocks.websocket.serverSend.mockClear();
await sut.getSegment(auth, assetId, 'session-a', variantIndex, 'init.mp4');
await sut.getSegment(auth, assetId, 'session-b', variantIndex, 'init.mp4');
expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsHeartbeat', {
sessionId: 'session-a',
variantIndex,
segmentIndex: 6,
});
expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsHeartbeat', {
sessionId: 'session-b',
variantIndex,
segmentIndex: 3,
});
});
it('rejects pending waiters for the previous variant on variant change', async () => {
mocks.storage.checkFileExists.mockResolvedValueOnce(false);
const pending = sut.getSegment(auth, assetId, sessionId, 0, 'seg_1.m4s');
await new Promise((resolve) => setImmediate(resolve));
await sut.getSegment(auth, assetId, sessionId, 1, 'seg_1.m4s');
await expect(pending).rejects.toThrow('Variant changed');
});
it('throws NotFoundException when the session does not exist', async () => {
mocks.videoStream.getSession.mockReset();
await expect(sut.getSegment(auth, assetId, sessionId, variantIndex, 'init.mp4')).rejects.toBeInstanceOf(
NotFoundException,
);
});
});
describe('endSession', () => {
it('emits HlsSessionEnd', async () => {
const auth = factory.auth();
const assetId = 'asset-1';
mocks.access.asset.checkOwnerAccess.mockResolvedValue(new Set([assetId]));
await sut.endSession(auth, assetId, sessionId);
expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsSessionEnd', { sessionId });
});
});
});
+198
View File
@@ -0,0 +1,198 @@
import { BadRequestException, Injectable, NotFoundException } from '@nestjs/common';
import { constants } from 'node:fs';
import { join } from 'node:path';
import {
HLS_SEGMENT_DURATION,
HLS_SEGMENT_FILENAME_REGEX,
HLS_VARIANTS,
HLS_VERSION,
SUPPORTED_HWA_CODECS,
} from 'src/constants';
import { StorageCore } from 'src/cores/storage.core';
import { OnEvent } from 'src/decorators';
import { AuthDto } from 'src/dtos/auth.dto';
import { SystemConfigFFmpegDto } from 'src/dtos/system-config.dto';
import { CacheControl, ImmichWorker, Permission } from 'src/enum';
import { ArgOf } from 'src/repositories/event.repository';
import { BaseService } from 'src/services/base.service';
import { VideoPacketInfo, VideoStreamInfo } from 'src/types';
import { PendingEvents } from 'src/utils/event';
import { ImmichFileResponse } from 'src/utils/file';
import { getOutputSize } from 'src/utils/media';
type AssetWithStreamInfo = { videoStream: VideoStreamInfo & { timeBase: number }; packets: VideoPacketInfo };
type ApiSession = { lastRequestedSegment: number | null; lastVariantIndex: number | null };
@Injectable()
export class HlsService extends BaseService {
private pendingSegments = new PendingEvents<'HlsSegmentResult'>({ timeoutMs: 15_000 });
private pendingSessions = new PendingEvents<'HlsSessionResult'>({ timeoutMs: 5000 });
private sessions = new Map<string, ApiSession>();
@OnEvent({ name: 'HlsSessionResult', server: true, workers: [ImmichWorker.Api] })
onSessionResult(event: ArgOf<'HlsSessionResult'>) {
this.pendingSessions.complete(event.sessionId, event);
if (event.error) {
this.sessions.delete(event.sessionId);
this.pendingSegments.rejectByPrefix(`${event.sessionId}:`, event.error);
}
}
@OnEvent({ name: 'HlsSessionEnd', server: true, workers: [ImmichWorker.Api] })
onSessionEnd({ sessionId }: ArgOf<'HlsSessionEnd'>) {
this.sessions.delete(sessionId);
this.pendingSegments.rejectByPrefix(`${sessionId}:`, 'Session ended');
}
@OnEvent({ name: 'HlsSegmentResult', server: true, workers: [ImmichWorker.Api] })
onSegmentResult(event: ArgOf<'HlsSegmentResult'>) {
this.pendingSegments.complete(this.getSegmentKey(event), event);
}
async getMainPlaylist(auth: AuthDto, assetId: string) {
await this.requireAccess({ auth, permission: Permission.AssetView, ids: [assetId] });
const { ffmpeg } = await this.getConfig({ withCache: true });
if (!ffmpeg.realtime.enabled) {
throw new BadRequestException('Real-time transcoding is not enabled');
}
const asset = await this.videoStreamRepository.getForMainPlaylist(assetId);
if (!asset) {
throw new NotFoundException('Asset is not yet ready for streaming');
}
// Sharing the sessionId allows only one microservices worker to successfully insert to the session table.
// The microservices worker that creates a session owns the transcoding lifecycle for it.
const sessionId = this.cryptoRepository.randomUUID();
this.websocketRepository.serverSend('HlsSessionRequest', { sessionId, assetId, ownerId: auth.user.id });
await this.pendingSessions.wait(sessionId);
this.trackSession(sessionId);
return this.generateMainPlaylist(sessionId, ffmpeg, asset);
}
async getMediaPlaylist(auth: AuthDto, assetId: string, sessionId: string) {
await this.requireAccess({ auth, permission: Permission.AssetView, ids: [assetId] });
const asset = await this.videoStreamRepository.getForMediaPlaylist(assetId, sessionId);
if (!asset) {
throw new NotFoundException('Asset not found or not yet ready for streaming');
}
return this.generateMediaPlaylist(asset);
}
async getSegment(auth: AuthDto, assetId: string, sessionId: string, variantIndex: number, filename: string) {
await this.requireAccess({ auth, permission: Permission.AssetView, ids: [assetId] });
const session = await this.videoStreamRepository.getSession(sessionId);
if (!session) {
throw new NotFoundException('Session not found');
}
const variantDir = StorageCore.getHlsVariantFolder({ ownerId: auth.user.id, sessionId, variantIndex });
const path = join(variantDir, filename);
const response = new ImmichFileResponse({
path,
contentType: 'video/mp4',
cacheControl: CacheControl.PrivateWithCache,
});
const apiSession = this.trackSession(sessionId, variantIndex);
const segmentIndex = this.getSegmentIndex(apiSession, filename);
this.websocketRepository.serverSend('HlsHeartbeat', { sessionId, variantIndex, segmentIndex });
if (await this.storageRepository.checkFileExists(path, constants.R_OK)) {
return response;
}
this.websocketRepository.serverSend('HlsSegmentRequest', { sessionId, assetId, variantIndex, segmentIndex });
await this.pendingSegments.wait(this.getSegmentKey({ sessionId, variantIndex, segmentIndex }));
return response;
}
async endSession(auth: AuthDto, assetId: string, sessionId: string): Promise<void> {
await this.requireAccess({ auth, permission: Permission.AssetView, ids: [assetId] });
this.websocketRepository.serverSend('HlsSessionEnd', { sessionId });
}
private generateMainPlaylist(sessionId: string, ffmpeg: SystemConfigFFmpegDto, asset: AssetWithStreamInfo) {
const fps = ((asset.packets.packetCount * asset.videoStream.timeBase) / asset.packets.totalDuration).toFixed(3);
const sourceResolution = Math.min(asset.videoStream.height, asset.videoStream.width);
const targetResolution = Math.max(sourceResolution, HLS_VARIANTS[0].resolution);
const lines = ['#EXTM3U', `#EXT-X-VERSION:${HLS_VERSION}`];
for (let i = 0; i < HLS_VARIANTS.length; i++) {
const { resolution, bitrate, codec, codecString } = HLS_VARIANTS[i];
if (resolution > targetResolution || !SUPPORTED_HWA_CODECS[ffmpeg.accel].includes(codec)) {
continue;
}
const { width, height } = getOutputSize(asset.videoStream, resolution);
lines.push(
`#EXT-X-STREAM-INF:BANDWIDTH=${bitrate},RESOLUTION=${width}x${height},CODECS="${codecString},mp4a.40.2",VIDEO-RANGE=SDR,FRAME-RATE=${fps}`,
`${sessionId}/${i}/playlist.m3u8`,
);
}
lines.push('');
if (lines.length === 3) {
throw new NotFoundException('No supported variants for this video');
}
return lines.join('\n');
}
private generateMediaPlaylist({ videoStream, packets }: AssetWithStreamInfo) {
const fps = (packets.packetCount * videoStream.timeBase) / packets.totalDuration;
const framesPerSegment = Math.ceil(HLS_SEGMENT_DURATION * fps);
const fullSegmentDuration = framesPerSegment / fps;
const segmentCount = Math.ceil(packets.outputFrames / framesPerSegment);
const lastSegmentFrames = packets.outputFrames - framesPerSegment * (segmentCount - 1);
const lastSegmentDuration = lastSegmentFrames / fps;
const lines = [
'#EXTM3U',
`#EXT-X-VERSION:${HLS_VERSION}`,
`#EXT-X-TARGETDURATION:${HLS_SEGMENT_DURATION}`,
'#EXT-X-MEDIA-SEQUENCE:0',
'#EXT-X-PLAYLIST-TYPE:VOD',
'#EXT-X-MAP:URI="init.mp4"',
];
for (let i = 0; i < segmentCount - 1; i++) {
lines.push(`#EXTINF:${fullSegmentDuration.toFixed(6)},`, `seg_${i}.m4s`);
}
lines.push(`#EXTINF:${lastSegmentDuration.toFixed(6)},`, `seg_${segmentCount - 1}.m4s`, '#EXT-X-ENDLIST', '');
return lines.join('\n');
}
private getSegmentKey({ sessionId, variantIndex, segmentIndex }: ArgOf<'HlsSegmentResult'>) {
return `${sessionId}:${variantIndex}:${segmentIndex}`;
}
private getSegmentIndex(session: ApiSession, filename: string) {
if (filename.endsWith('.mp4')) {
return (session.lastRequestedSegment ?? -1) + 1;
}
const segmentIndex = Number.parseInt(HLS_SEGMENT_FILENAME_REGEX.exec(filename)![1]);
session.lastRequestedSegment = segmentIndex;
return segmentIndex;
}
private trackSession(id: string, variantIndex: number | null = null) {
const session = this.sessions.get(id);
if (!session) {
const newSession = { lastRequestedSegment: null, lastVariantIndex: variantIndex };
this.sessions.set(id, newSession);
return newSession;
}
if (session.lastVariantIndex !== null && session.lastVariantIndex !== variantIndex) {
this.pendingSegments.rejectByPrefix(`${id}:${session.lastVariantIndex}:`, 'Variant changed');
}
session.lastVariantIndex = variantIndex;
return session;
}
}
+4
View File
@@ -11,6 +11,7 @@ import { DatabaseBackupService } from 'src/services/database-backup.service';
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 { JobService } from 'src/services/job.service';
import { LibraryService } from 'src/services/library.service';
import { MaintenanceService } from 'src/services/maintenance.service';
@@ -39,6 +40,7 @@ import { SystemMetadataService } from 'src/services/system-metadata.service';
import { TagService } from 'src/services/tag.service';
import { TelemetryService } from 'src/services/telemetry.service';
import { TimelineService } from 'src/services/timeline.service';
import { TranscodingService } from 'src/services/transcoding.service';
import { TrashService } from 'src/services/trash.service';
import { UserAdminService } from 'src/services/user-admin.service';
import { UserService } from 'src/services/user.service';
@@ -61,6 +63,7 @@ export const services = [
DatabaseService,
DownloadService,
DuplicateService,
HlsService,
JobService,
LibraryService,
MaintenanceService,
@@ -89,6 +92,7 @@ export const services = [
TagService,
TelemetryService,
TimelineService,
TranscodingService,
TrashService,
UserAdminService,
UserService,
+24 -2
View File
@@ -57,7 +57,6 @@ describe(PersonService.name, () => {
],
});
expect(mocks.person.getAllForUser).toHaveBeenCalledWith({ skip: 0, take: 10 }, auth.user.id, {
minimumFaceCount: 3,
withHidden: true,
});
});
@@ -84,7 +83,6 @@ describe(PersonService.name, () => {
],
});
expect(mocks.person.getAllForUser).toHaveBeenCalledWith({ skip: 0, take: 10 }, auth.user.id, {
minimumFaceCount: 3,
withHidden: false,
});
});
@@ -454,6 +452,30 @@ describe(PersonService.name, () => {
expect(mocks.person.update).not.toHaveBeenCalled();
expect(mocks.job.queueAll).not.toHaveBeenCalled();
});
it('should reject creating a face on an asset the user does not own', async () => {
const auth = AuthFactory.create();
const asset = AssetFactory.create();
const person = PersonFactory.create({ faceAssetId: null });
mocks.access.asset.checkOwnerAccess.mockResolvedValue(new Set());
mocks.access.person.checkOwnerAccess.mockResolvedValue(new Set([person.id]));
await expect(
sut.createFace(auth, {
assetId: asset.id,
personId: person.id,
imageHeight: 500,
imageWidth: 400,
x: 10,
y: 20,
width: 100,
height: 110,
}),
).rejects.toBeInstanceOf(BadRequestException);
expect(mocks.person.createAssetFace).not.toHaveBeenCalled();
});
});
describe('createNewFeaturePhoto', () => {
+1 -3
View File
@@ -63,9 +63,7 @@ export class PersonService extends BaseService {
}
closestFaceAssetId = person.faceAssetId;
}
const { machineLearning } = await this.getConfig({ withCache: false });
const { items, hasNextPage } = await this.personRepository.getAllForUser(pagination, auth.user.id, {
minimumFaceCount: machineLearning.facialRecognition.minFaces,
withHidden,
closestFaceAssetId,
});
@@ -627,7 +625,7 @@ export class PersonService extends BaseService {
// TODO return a asset face response
async createFace(auth: AuthDto, dto: AssetFaceCreateDto): Promise<void> {
await Promise.all([
this.requireAccess({ auth, permission: Permission.AssetRead, ids: [dto.assetId] }),
this.requireAccess({ auth, permission: Permission.AssetUpdate, ids: [dto.assetId] }),
this.requireAccess({ auth, permission: Permission.PersonRead, ids: [dto.personId] }),
]);
@@ -41,6 +41,7 @@ describe(QueueService.name, () => {
{ name: JobName.PersonCleanup },
{ name: JobName.MemoryCleanup },
{ name: JobName.SessionCleanup },
{ name: JobName.HlsSessionCleanup },
{ name: JobName.AuditTableCleanup },
{ name: JobName.MemoryGenerate },
{ name: JobName.UserSyncUsage },
+1
View File
@@ -269,6 +269,7 @@ export class QueueService extends BaseService {
{ name: JobName.PersonCleanup },
{ name: JobName.MemoryCleanup },
{ name: JobName.SessionCleanup },
{ name: JobName.HlsSessionCleanup },
{ name: JobName.AuditTableCleanup },
);
}
@@ -148,6 +148,7 @@ describe(ServerService.name, () => {
configFile: false,
trash: true,
email: false,
realtimeTranscoding: false,
});
expect(mocks.systemMetadata.get).toHaveBeenCalled();
});
@@ -167,6 +168,7 @@ describe(ServerService.name, () => {
mapDarkStyleUrl: 'https://tiles.immich.cloud/v1/style/dark.json',
mapLightStyleUrl: 'https://tiles.immich.cloud/v1/style/light.json',
maintenanceMode: false,
minFaces: 3,
});
expect(mocks.systemMetadata.get).toHaveBeenCalled();
});
+3 -1
View File
@@ -86,7 +86,7 @@ export class ServerService extends BaseService {
}
async getFeatures(): Promise<ServerFeaturesDto> {
const { reverseGeocoding, metadata, map, machineLearning, trash, oauth, passwordLogin, notifications } =
const { reverseGeocoding, metadata, map, machineLearning, trash, oauth, passwordLogin, notifications, ffmpeg } =
await this.getConfig({ withCache: false });
const { configFile } = this.configRepository.getEnv();
@@ -106,6 +106,7 @@ export class ServerService extends BaseService {
passwordLogin: passwordLogin.enabled,
configFile: !!configFile,
email: notifications.smtp.enabled,
realtimeTranscoding: ffmpeg.realtime.enabled,
};
}
@@ -127,6 +128,7 @@ export class ServerService extends BaseService {
mapDarkStyleUrl: config.map.darkStyle,
mapLightStyleUrl: config.map.lightStyle,
maintenanceMode: false,
minFaces: config.machineLearning.facialRecognition.minFaces,
};
}
@@ -1,5 +1,6 @@
import { BadRequestException } from '@nestjs/common';
import { defaults, SystemConfig } from 'src/config';
import { ReleaseChannel } from 'src/dtos/system-config.dto';
import {
AudioCodec,
Colorspace,
@@ -72,6 +73,9 @@ const updatedConfig = Object.freeze<SystemConfig>({
accel: TranscodeHardwareAcceleration.Disabled,
accelDecode: true,
tonemap: ToneMapping.Hable,
realtime: {
enabled: false,
},
},
logging: {
enabled: true,
@@ -184,6 +188,7 @@ const updatedConfig = Object.freeze<SystemConfig>({
},
newVersionCheck: {
enabled: true,
channel: ReleaseChannel.Stable,
},
trash: {
enabled: true,
@@ -0,0 +1,539 @@
import {
HLS_BACKPRESSURE_PAUSE_SEGMENTS,
HLS_BACKPRESSURE_RESUME_SEGMENTS,
HLS_CLEANUP_INTERVAL_MS,
HLS_INACTIVITY_TIMEOUT_MS,
HLS_LEASE_DURATION_MS,
} from 'src/constants';
import { TranscodingService } from 'src/services/transcoding.service';
import { VIDEO_STREAM_SESSION_PK_CONSTRAINT } from 'src/utils/database';
import { eiffelTower, train, waterfall } from 'test/fixtures/media.stub';
import { mockSpawn, newTestService, ServiceMocks } from 'test/utils';
import { vi } from 'vitest';
describe(TranscodingService.name, () => {
let sut: TranscodingService;
let mocks: ServiceMocks;
const sessionId = 'session-1';
const assetId = 'asset-1';
const ownerId = 'user-1';
const completeSegment = (index: number) => {
const listener = vi.mocked(mocks.storage.watchDir).mock.lastCall?.[1];
expect(listener).toBeDefined();
listener!('rename', `seg_${index}.m4s`);
};
const completeSegmentsThrough = (start: number, end: number) => {
for (let i = start; i <= end; i++) {
completeSegment(i);
}
};
beforeEach(() => {
({ sut, mocks } = newTestService(TranscodingService));
mocks.systemMetadata.get.mockResolvedValue({ ffmpeg: { realtime: { enabled: true } } });
mocks.videoStream.getForTranscoding.mockResolvedValue(eiffelTower);
});
describe('onSessionRequest', () => {
it('creates the session row and emits HlsSessionResult on success', async () => {
await sut.onSessionRequest({ sessionId, assetId, ownerId });
expect(mocks.videoStream.createSession).toHaveBeenCalledWith({
id: sessionId,
assetId,
expiresAt: expect.any(Date),
});
expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsSessionResult', { sessionId });
});
it('treats a primary-key conflict as a no-op for replay tolerance', async () => {
mocks.videoStream.createSession.mockRejectedValue({ constraint_name: VIDEO_STREAM_SESSION_PK_CONSTRAINT });
await sut.onSessionRequest({ sessionId, assetId, ownerId });
expect(mocks.websocket.serverSend).not.toHaveBeenCalled();
});
it('emits HlsSessionResult with an error on other DB failures', async () => {
mocks.videoStream.createSession.mockRejectedValue(new Error('database is down'));
await sut.onSessionRequest({ sessionId, assetId, ownerId });
expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsSessionResult', {
sessionId,
error: 'Failed to create HLS session',
});
});
});
describe('onSessionEnd', () => {
it('removes the session, kills the transcode, and deletes the dir + DB row', async () => {
await sut.onSessionRequest({ sessionId, assetId, ownerId });
const process = mockSpawn(0, '', '');
mocks.process.spawn.mockReturnValue(process);
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex: 0 });
await sut.onSessionEnd({ sessionId });
expect(process.kill).toHaveBeenCalled();
expect(mocks.storage.unlinkDir).toHaveBeenCalled();
expect(mocks.videoStream.deleteSession).toHaveBeenCalledWith(sessionId);
});
it('is a no-op when the session is unknown', async () => {
await sut.onSessionEnd({ sessionId: 'never-created' });
expect(mocks.videoStream.deleteSession).not.toHaveBeenCalled();
expect(mocks.storage.unlinkDir).not.toHaveBeenCalled();
});
});
describe('onHeartbeat', () => {
it('extends the DB lease when remaining time falls below half', async () => {
vi.useFakeTimers();
try {
await sut.onSessionRequest({ sessionId, assetId, ownerId });
vi.setSystemTime(Date.now() + HLS_LEASE_DURATION_MS / 2 + 1);
await sut.onHeartbeat({ sessionId });
expect(mocks.videoStream.extendSession).toHaveBeenCalledWith(sessionId, expect.any(Date));
} finally {
vi.useRealTimers();
}
});
it('does not extend the lease while it is still fresh', async () => {
await sut.onSessionRequest({ sessionId, assetId, ownerId });
await sut.onHeartbeat({ sessionId });
expect(mocks.videoStream.extendSession).not.toHaveBeenCalled();
});
it('is a no-op when the session is unknown', async () => {
await sut.onHeartbeat({ sessionId: 'never-created' });
expect(mocks.videoStream.extendSession).not.toHaveBeenCalled();
});
});
describe('onSegmentRequest', () => {
beforeEach(async () => {
await sut.onSessionRequest({ sessionId, assetId, ownerId });
mocks.websocket.serverSend.mockClear();
});
it('spawns FFmpeg on the first request', async () => {
mocks.process.spawn.mockReturnValue(mockSpawn(0, '', ''));
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex: 0 });
expect(mocks.process.spawn).toHaveBeenCalledTimes(1);
expect(mocks.process.spawn).toHaveBeenCalledWith('ffmpeg', expect.any(Array), expect.any(Object));
});
it('kills and respawns when the variant changes', async () => {
const first = mockSpawn(0, '', '');
const second = mockSpawn(0, '', '');
mocks.process.spawn.mockReturnValueOnce(first).mockReturnValueOnce(second);
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex: 0 });
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 1, segmentIndex: 0 });
expect(first.kill).toHaveBeenCalled();
expect(mocks.process.spawn).toHaveBeenCalledTimes(2);
});
it('kills and respawns when seeking before the start segment', async () => {
const first = mockSpawn(0, '', '');
const second = mockSpawn(0, '', '');
mocks.process.spawn.mockReturnValueOnce(first).mockReturnValueOnce(second);
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex: 5 });
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex: 2 });
expect(first.kill).toHaveBeenCalled();
expect(mocks.process.spawn).toHaveBeenCalledTimes(2);
});
it('kills and respawns when the requested segment is too far ahead', async () => {
const first = mockSpawn(0, '', '');
const second = mockSpawn(0, '', '');
mocks.process.spawn.mockReturnValueOnce(first).mockReturnValueOnce(second);
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex: 0 });
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex: 5 });
expect(first.kill).toHaveBeenCalled();
expect(mocks.process.spawn).toHaveBeenCalledTimes(2);
});
it('does not spawn when the session is unknown', async () => {
await sut.onSegmentRequest({ sessionId: 'never-created', assetId, variantIndex: 0, segmentIndex: 0 });
expect(mocks.process.spawn).not.toHaveBeenCalled();
});
it('accepts segments from a restart after the previous ffmpeg exited on its own', async () => {
const first = mockSpawn(0, '', '');
const second = mockSpawn(0, '', '');
mocks.process.spawn.mockReturnValueOnce(first).mockReturnValueOnce(second);
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex: 10 });
completeSegment(10);
const onCalls = vi.mocked(first.on).mock.calls as unknown as [string, (code: number) => void][];
const exitHandler = onCalls.find(([event]) => event === 'exit')?.[1];
exitHandler?.(0);
mocks.websocket.serverSend.mockClear();
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex: 2 });
completeSegment(2);
expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsSegmentResult', {
sessionId,
variantIndex: 0,
segmentIndex: 2,
});
});
});
describe('backpressure', () => {
let proc: ReturnType<typeof mockSpawn>;
beforeEach(async () => {
proc = mockSpawn(0, '', '');
mocks.process.spawn.mockReturnValue(proc);
await sut.onSessionRequest({ sessionId, assetId, ownerId });
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex: 0 });
});
it('pauses the transcode once the lead exceeds HLS_BACKPRESSURE_PAUSE_SEGMENTS', async () => {
completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1);
await sut.onHeartbeat({ sessionId, segmentIndex: 0 });
expect(proc.kill).toHaveBeenCalledWith('SIGSTOP');
});
it('does not pause when the lead equals the pause threshold', async () => {
completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS);
await sut.onHeartbeat({ sessionId, segmentIndex: 0 });
expect(proc.kill).not.toHaveBeenCalled();
});
it('resumes once the lead drops below HLS_BACKPRESSURE_RESUME_SEGMENTS', async () => {
completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1);
await sut.onHeartbeat({ sessionId, segmentIndex: 0 });
expect(proc.kill).toHaveBeenCalledWith('SIGSTOP');
vi.mocked(proc.kill).mockClear();
const requested = HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1 - (HLS_BACKPRESSURE_RESUME_SEGMENTS - 1);
await sut.onHeartbeat({ sessionId, segmentIndex: requested });
expect(proc.kill).toHaveBeenCalledWith('SIGCONT');
});
it('stays paused while the lead is in the dead-band', async () => {
completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1);
await sut.onHeartbeat({ sessionId, segmentIndex: 0 });
vi.mocked(proc.kill).mockClear();
const requested = HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1 - HLS_BACKPRESSURE_RESUME_SEGMENTS;
await sut.onHeartbeat({ sessionId, segmentIndex: requested });
expect(proc.kill).not.toHaveBeenCalled();
});
it('is a no-op when no segment has completed yet', async () => {
await sut.onHeartbeat({ sessionId, segmentIndex: 0 });
expect(proc.kill).not.toHaveBeenCalled();
});
it('is a no-op when the heartbeat omits segmentIndex', async () => {
completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1);
await sut.onHeartbeat({ sessionId });
expect(proc.kill).not.toHaveBeenCalled();
});
it('resumes the paused transcode when the client requests the next in-range segment', async () => {
completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1);
await sut.onHeartbeat({ sessionId, segmentIndex: 0 });
expect(proc.kill).toHaveBeenCalledWith('SIGSTOP');
vi.mocked(proc.kill).mockClear();
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex: 1 });
expect(proc.kill).toHaveBeenCalledWith('SIGCONT');
expect(mocks.process.spawn).toHaveBeenCalledTimes(1);
});
it('does not re-pause a freshly spawned transcode after a seek-driven restart', async () => {
const newProc = mockSpawn(0, '', '');
mocks.process.spawn.mockReturnValueOnce(newProc);
completeSegmentsThrough(0, HLS_BACKPRESSURE_PAUSE_SEGMENTS + 1);
await sut.onHeartbeat({ sessionId, segmentIndex: 0 });
expect(proc.kill).toHaveBeenCalledWith('SIGSTOP');
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 1, segmentIndex: 0 });
vi.mocked(newProc.kill).mockClear();
await sut.onHeartbeat({ sessionId, segmentIndex: 0 });
expect(newProc.kill).not.toHaveBeenCalled();
});
it('ignores stale segment events from the prior transcode after a backward seek', async () => {
const newProc = mockSpawn(0, '', '');
mocks.process.spawn.mockReturnValueOnce(newProc);
const completedAhead = HLS_BACKPRESSURE_PAUSE_SEGMENTS + 5;
completeSegmentsThrough(1, completedAhead); // seg_0 was emitted in beforeEach
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 1, segmentIndex: 0 });
vi.mocked(newProc.kill).mockClear();
mocks.websocket.serverSend.mockClear();
completeSegment(completedAhead + 1);
expect(mocks.websocket.serverSend).not.toHaveBeenCalledWith(
'HlsSegmentResult',
expect.objectContaining({ segmentIndex: completedAhead + 1 }),
);
expect(newProc.kill).not.toHaveBeenCalled();
completeSegment(0);
expect(mocks.websocket.serverSend).toHaveBeenCalledWith(
'HlsSegmentResult',
expect.objectContaining({ segmentIndex: 0 }),
);
});
});
describe('inactivity sweeper', () => {
it('reaps a session whose last activity exceeds the inactivity timeout', async () => {
vi.useFakeTimers();
try {
await sut.onSessionRequest({ sessionId, assetId, ownerId });
mocks.websocket.serverSend.mockClear();
await vi.advanceTimersByTimeAsync(HLS_INACTIVITY_TIMEOUT_MS + HLS_CLEANUP_INTERVAL_MS);
expect(mocks.websocket.serverSend).toHaveBeenCalledWith('HlsSessionEnd', { sessionId });
expect(mocks.videoStream.deleteSession).toHaveBeenCalledWith(sessionId);
} finally {
vi.useRealTimers();
}
});
});
describe('onShutdown', () => {
it('ends every active session', async () => {
await sut.onSessionRequest({ sessionId: 'session-a', assetId, ownerId });
await sut.onSessionRequest({ sessionId: 'session-b', assetId, ownerId });
await sut.onShutdown();
expect(mocks.videoStream.deleteSession).toHaveBeenCalledWith('session-a');
expect(mocks.videoStream.deleteSession).toHaveBeenCalledWith('session-b');
});
});
describe('onHlsSessionCleanup', () => {
it('reaps DB-expired sessions under a database lock', async () => {
mocks.database.withLock.mockImplementation(async (_, fn) => fn());
mocks.videoStream.getExpiredSessions.mockResolvedValue([
{ id: 'expired-1', ownerId: 'user-a' },
{ id: 'expired-2', ownerId: 'user-b' },
]);
await sut.onHlsSessionCleanup();
expect(mocks.videoStream.deleteSession).toHaveBeenCalledWith('expired-1');
expect(mocks.videoStream.deleteSession).toHaveBeenCalledWith('expired-2');
expect(mocks.storage.unlinkDir).toHaveBeenCalledTimes(2);
});
});
describe('FFmpeg full command', () => {
const baseCommand = [
'-nostdin',
'-nostats',
'-i',
'eiffel-tower.mp4',
'-map',
'0:0',
'-map_metadata',
'-1',
'-map',
'0:1',
'-g',
'50',
'-keyint_min',
'50',
'-crf',
'23',
'-copyts',
'-r',
'50130000/2012441',
'-avoid_negative_ts',
'disabled',
'-f',
'hls',
'-hls_time',
'2',
'-hls_list_size',
'0',
'-hls_segment_type',
'fmp4',
'-hls_fmp4_init_filename',
'init.mp4',
'-hls_segment_options',
'movflags=+frag_discont',
'-hls_flags',
'temp_file',
'-start_number',
'0',
];
it.each([
{
variantIndex: 6,
expected: [
...baseCommand,
'-c:v',
'libsvtav1',
'-c:a',
'aac',
'-preset',
'12',
'-svtav1-params',
'hierarchical-levels=3:lookahead=0:enable-tf=0:mbr=4000k',
'-hls_segment_filename',
'/data/encoded-video/user-1/se/ss/session-1/6/seg_%d.m4s',
'/data/encoded-video/user-1/se/ss/session-1/6/playlist.m3u8',
].sort(),
},
{
variantIndex: 4,
expected: [
...baseCommand,
'-c:v',
'hevc',
'-c:a',
'aac',
'-tag:v',
'hvc1',
'-preset',
'ultrafast',
'-maxrate',
'2500k',
'-bufsize',
'5000k',
'-x265-params',
'no-scenecut=1:no-open-gop=1',
'-vf',
'scale=720:-2',
'-hls_segment_filename',
'/data/encoded-video/user-1/se/ss/session-1/4/seg_%d.m4s',
'/data/encoded-video/user-1/se/ss/session-1/4/playlist.m3u8',
].sort(),
},
{
variantIndex: 2,
expected: [
...baseCommand,
'-c:v',
'h264',
'-c:a',
'aac',
'-preset',
'ultrafast',
'-maxrate',
'2500k',
'-bufsize',
'5000k',
'-sc_threshold:v',
'0',
'-vf',
'scale=480:-2',
'-hls_segment_filename',
'/data/encoded-video/user-1/se/ss/session-1/2/seg_%d.m4s',
'/data/encoded-video/user-1/se/ss/session-1/2/playlist.m3u8',
].sort(),
},
])('builds the expected FFmpeg command for $codec (variant $variantIndex)', async ({ variantIndex, expected }) => {
mocks.process.spawn.mockReturnValue(mockSpawn(0, '', ''));
await sut.onSessionRequest({ sessionId, assetId, ownerId });
await sut.onSegmentRequest({ sessionId, assetId, variantIndex, segmentIndex: 0 });
expect(mocks.process.spawn.mock.calls[0][1].toSorted()).toEqual(expected);
});
});
describe('FFmpeg seek per segment', () => {
const eiffelSeeks = [
0, 1.987_15, 3.994_372_222_222_222, 6.001_594_444_444_444, 8.008_816_666_666_666, 10.016_038_888_888_888,
12.023_261_111_111_111, 14.030_483_333_333_333, 16.037_705_555_555_554, 18.044_927_777_777_776,
20.052_149_999_999_997, 22.059_372_222_222_223,
];
const waterfallSeeks = [
0, 1.994_642_826_321_467, 4.006_047_357_065_803, 6.017_451_887_810_139_5, 8.028_856_418_554_476,
10.040_260_949_298_812,
];
const trainSeeks = [
0, 1.991_666_666_666_666_7, 3.991_666_666_666_666_7, 5.991_666_666_666_666, 7.991_666_666_666_666,
9.991_666_666_666_667, 11.991_666_666_666_667, 13.991_666_666_666_667, 15.991_666_666_666_667,
17.991_666_666_666_667, 19.991_666_666_666_667,
];
const cases = [
...eiffelSeeks.map((expected, segmentIndex) => ({
name: `${eiffelTower.originalPath} K=${segmentIndex}`,
fixture: eiffelTower,
segmentIndex,
expected,
})),
...waterfallSeeks.map((expected, segmentIndex) => ({
name: `${waterfall.originalPath} K=${segmentIndex}`,
fixture: waterfall,
segmentIndex,
expected,
})),
...trainSeeks.map((expected, segmentIndex) => ({
name: `${train.originalPath} K=${segmentIndex}`,
fixture: train,
segmentIndex,
expected,
})),
];
it.each(cases)('$name', async ({ fixture, segmentIndex, expected }) => {
mocks.videoStream.getForTranscoding.mockResolvedValue(fixture);
mocks.process.spawn.mockReturnValue(mockSpawn(0, '', ''));
await sut.onSessionRequest({ sessionId, assetId, ownerId });
await sut.onSegmentRequest({ sessionId, assetId, variantIndex: 0, segmentIndex });
const args = mocks.process.spawn.mock.calls[0][1] as string[];
if (expected === 0) {
expect(args).toEqual(expect.arrayContaining(['-copyts', '-avoid_negative_ts', 'disabled']));
expect(args).not.toContain('-ss');
} else {
expect(args).toEqual(
expect.arrayContaining(['-ss', String(expected), '-copyts', '-avoid_negative_ts', 'disabled']),
);
}
});
});
});
+387
View File
@@ -0,0 +1,387 @@
import { Injectable } from '@nestjs/common';
import { ChildProcess } from 'node:child_process';
import { join } from 'node:path';
import {
HLS_BACKPRESSURE_PAUSE_SEGMENTS,
HLS_BACKPRESSURE_RESUME_SEGMENTS,
HLS_CLEANUP_INTERVAL_MS,
HLS_INACTIVITY_TIMEOUT_MS,
HLS_LEASE_DURATION_MS,
HLS_SEGMENT_DURATION,
HLS_SEGMENT_FILENAME_REGEX,
HLS_VARIANTS,
} from 'src/constants';
import { StorageCore } from 'src/cores/storage.core';
import { OnEvent, OnJob } from 'src/decorators';
import { DatabaseLock, ImmichWorker, JobName, QueueName, TranscodeTarget } from 'src/enum';
import { ArgOf } from 'src/repositories/event.repository';
import { BaseService } from 'src/services/base.service';
import { VideoInterfaces } from 'src/types';
import { isVideoStreamSessionPkConstraint } from 'src/utils/database';
import { BaseConfig } from 'src/utils/media';
type Session = {
assetId: string;
expiresAt: Date;
id: string;
lastActivityTime: Date;
lastClientRequestedSegment: number | null;
lastCompletedSegment: number | null;
ownerId: string;
paused: boolean;
process: ChildProcess | null;
startSegment: number | null;
variantIndex: number | null;
};
@Injectable()
export class TranscodingService extends BaseService {
private sessions = new Map<string, Session>();
private videoInterfaces: VideoInterfaces = { dri: [], mali: false };
private cleanupInterval: NodeJS.Timeout | null = null;
@OnEvent({ name: 'AppBootstrap', workers: [ImmichWorker.Microservices] })
async onBootstrap() {
const [videoInterfaces] = await Promise.all([this.storageCore.getVideoInterfaces(), this.removeExpiredSessions()]);
this.videoInterfaces = videoInterfaces;
}
@OnEvent({ name: 'AppShutdown', workers: [ImmichWorker.Microservices] })
onShutdown() {
if (this.cleanupInterval) {
clearInterval(this.cleanupInterval);
this.cleanupInterval = null;
}
return Promise.all([...this.sessions.values()].map(({ id }) => this.onSessionEnd({ sessionId: id })));
}
@OnJob({ name: JobName.HlsSessionCleanup, queue: QueueName.BackgroundTask })
onHlsSessionCleanup() {
return this.removeExpiredSessions();
}
@OnEvent({ name: 'HlsSessionRequest', server: true, workers: [ImmichWorker.Microservices] })
async onSessionRequest({ assetId, sessionId, ownerId }: ArgOf<'HlsSessionRequest'>) {
try {
const expiresAt = new Date(Date.now() + HLS_LEASE_DURATION_MS);
await this.videoStreamRepository.createSession({ id: sessionId, assetId, expiresAt });
this.sessions.set(sessionId, {
assetId,
expiresAt,
id: sessionId,
lastActivityTime: new Date(),
lastClientRequestedSegment: null,
lastCompletedSegment: null,
ownerId,
paused: false,
process: null,
startSegment: null,
variantIndex: null,
});
this.cleanupInterval ??= setInterval(() => void this.removeInactiveSessions(), HLS_CLEANUP_INTERVAL_MS);
this.websocketRepository.serverSend('HlsSessionResult', { sessionId });
} catch (error) {
// If insertion failed due to a PK constraint, another worker has already created a session for this ID.
if (!isVideoStreamSessionPkConstraint(error)) {
this.logger.error(`Failed to create HLS session ${sessionId}: ${error}`);
this.websocketRepository.serverSend('HlsSessionResult', { sessionId, error: 'Failed to create HLS session' });
}
}
}
@OnEvent({ name: 'HlsSessionEnd', server: true, workers: [ImmichWorker.Microservices] })
async onSessionEnd({ sessionId }: ArgOf<'HlsSessionEnd'>) {
const session = this.sessions.get(sessionId);
if (!session) {
return;
}
this.sessions.delete(sessionId);
if (this.cleanupInterval && this.sessions.size === 0) {
clearInterval(this.cleanupInterval);
this.cleanupInterval = null;
}
this.stopTranscode(session);
await this.removeSessionDir(session);
await this.videoStreamRepository.deleteSession(sessionId);
}
@OnEvent({ name: 'HlsHeartbeat', server: true, workers: [ImmichWorker.Microservices] })
async onHeartbeat({ sessionId, segmentIndex }: ArgOf<'HlsHeartbeat'>) {
const session = this.sessions.get(sessionId);
if (!session) {
return;
}
session.lastActivityTime = new Date();
if (segmentIndex !== undefined) {
session.lastClientRequestedSegment = segmentIndex;
this.applyBackpressure(session);
}
const remaining = session.expiresAt.getTime() - Date.now();
if (remaining < HLS_LEASE_DURATION_MS / 2) {
session.expiresAt = new Date(Date.now() + HLS_LEASE_DURATION_MS);
await this.videoStreamRepository.extendSession(sessionId, session.expiresAt);
}
}
@OnEvent({ name: 'HlsSegmentRequest', server: true, workers: [ImmichWorker.Microservices] })
async onSegmentRequest({ sessionId, variantIndex, segmentIndex }: ArgOf<'HlsSegmentRequest'>) {
const session = this.sessions.get(sessionId);
if (!session) {
return;
}
session.variantIndex ??= variantIndex;
session.startSegment ??= segmentIndex;
const curSegment = session.lastCompletedSegment === null ? session.startSegment : session.lastCompletedSegment + 1;
const needsRestart =
session.variantIndex !== variantIndex || segmentIndex < session.startSegment || segmentIndex > curSegment + 1;
if (needsRestart) {
this.stopTranscode(session);
session.variantIndex = variantIndex;
session.startSegment = segmentIndex;
} else if (session.process) {
this.resumeTranscode(session);
return;
}
const process = await this.startTranscode(session, variantIndex, segmentIndex);
if (process) {
session.process = process;
}
}
private applyBackpressure(session: Session) {
if (session.lastCompletedSegment === null || session.lastClientRequestedSegment === null) {
return;
}
const lead = session.lastCompletedSegment - session.lastClientRequestedSegment;
this.logger.debug(`Session ${session.id} lead is ${lead} segments`);
if (!session.paused && lead > HLS_BACKPRESSURE_PAUSE_SEGMENTS) {
this.pauseTranscode(session);
} else if (session.paused && lead < HLS_BACKPRESSURE_RESUME_SEGMENTS) {
this.resumeTranscode(session);
}
}
private async startTranscode(session: Session, variantIndex: number, startSegment: number) {
const { ffmpeg } = await this.getConfig({ withCache: true });
const asset = await this.videoStreamRepository.getForTranscoding(session.assetId);
if (!asset) {
this.logger.error(`Asset ${session.assetId} not found for HLS transcoding`);
return;
}
if (session.variantIndex !== variantIndex || session.startSegment !== startSegment) {
return;
}
const variant = HLS_VARIANTS[variantIndex];
if (!variant) {
this.logger.error(`Variant ${variantIndex} out of range for asset ${session.assetId}`);
await this.failSession(session, `Invalid variant index ${variantIndex}`);
return;
}
const variantDir = StorageCore.getHlsVariantFolder({
ownerId: session.ownerId,
sessionId: session.id,
variantIndex,
});
this.storageRepository.mkdirSync(variantDir);
// Encoder runs at fps = packetCount × timeBase / totalDuration with
// gop = ceil(SEGMENT_DURATION × fps). To start segment K's content at
// exactly cfr slot K × gop, seek to the midpoint between slots K×gop1 and
// K×gop. accurate_seek's "discard < target" then keeps the source frame
// that quantizes to slot K×gop and discards the one quantizing to K×gop1.
const fps = (asset.packets.packetCount * asset.videoStream.timeBase) / asset.packets.totalDuration;
const gop = Math.ceil(HLS_SEGMENT_DURATION * fps);
const seekSeconds = startSegment > 0 ? (startSegment * gop - 0.5) / fps : 0;
let config;
try {
config = BaseConfig.create(
{
...ffmpeg,
targetVideoCodec: variant.codec,
targetResolution: String(variant.resolution),
maxBitrate: `${Math.round(variant.bitrate / 1000)}k`,
gopSize: gop,
},
this.videoInterfaces,
{ strictGop: true, lowLatency: true },
);
} catch (error: any) {
this.logger.error(
`Failed to create transcode config for variant ${variantIndex} asset ${session.assetId}: ${error?.message ?? error}`,
);
await this.failSession(session, `Failed to start transcode: ${error?.message ?? 'unknown error'}`);
return;
}
const args = config.getHlsCommand(
{
initFilename: 'init.mp4',
inputPath: asset.originalPath,
packetCount: asset.packets.packetCount,
playlistFilename: join(variantDir, 'playlist.m3u8'),
seekSeconds,
segmentDuration: HLS_SEGMENT_DURATION,
segmentFilename: join(variantDir, 'seg_%d.m4s'),
startSegment,
target: TranscodeTarget.All,
timeBase: asset.videoStream.timeBase,
totalDuration: asset.packets.totalDuration,
},
asset.videoStream,
asset.audioStream ?? undefined,
);
this.logger.log(
`Starting HLS transcode for asset ${session.assetId} variant ${variantIndex} with command: ffmpeg ${args.join(' ')}`,
);
const process = this.processRepository.spawn('ffmpeg', args, { stdio: ['ignore', 'ignore', 'pipe'] });
this.attachProcessHandlers(process, session, variantIndex);
return process;
}
private failSession(session: Session, error: string) {
this.websocketRepository.serverSend('HlsSessionResult', { sessionId: session.id, error });
return this.onSessionEnd({ sessionId: session.id });
}
private attachProcessHandlers(process: ChildProcess, session: Session, variantIndex: number) {
let stderr = '';
const variantDir = StorageCore.getHlsVariantFolder({
ownerId: session.ownerId,
sessionId: session.id,
variantIndex,
});
// hlsenc writes each segment as `seg_K.m4s.tmp` then renames to
// `seg_K.m4s`. The rename event fires the moment the renamed file is
// observable — the only signal we need to tell the API worker the
// segment is ready to serve.
const watcher = this.storageRepository.watchDir(variantDir, (eventType, filename) => {
if (eventType !== 'rename' || !filename || session.process !== process) {
return;
}
const match = HLS_SEGMENT_FILENAME_REGEX.exec(filename);
if (!match) {
return;
}
const segmentIndex = Number.parseInt(match[1]);
const expected = session.lastCompletedSegment === null ? session.startSegment : session.lastCompletedSegment + 1;
// Ignore stale events from old process after seek
if (expected === null || segmentIndex !== expected) {
return;
}
session.lastCompletedSegment = segmentIndex;
this.websocketRepository.serverSend('HlsSegmentResult', {
sessionId: session.id,
variantIndex,
segmentIndex,
});
this.applyBackpressure(session);
});
watcher.on('error', (error) => {
this.logger.error(`watcher error for ${variantDir}: ${error}`);
});
process.stderr!.on('data', (chunk: Buffer) => {
if (session.process !== process) {
return;
}
stderr += chunk.toString();
});
process.on('exit', (code) => {
watcher.close();
if (session.process !== process || session.variantIndex !== variantIndex) {
return;
}
session.paused = false;
session.process = null;
session.lastCompletedSegment = null;
if (code) {
this.logger.error(
`FFmpeg exited with code ${code} for variant ${variantIndex} asset ${session.assetId}\n${stderr}`,
);
void this.failSession(session, `Transcoding process exited unexpectedly with code ${code}`).catch((error) =>
this.logger.error(`Failed to end session ${session.id} after ffmpeg exit: ${error}`),
);
}
});
}
private stopTranscode(session: Session) {
if (!session.process) {
return;
}
// SIGTERM makes it rename .tmp segments to .m4s even if they're still incomplete
session.process.kill('SIGKILL');
session.process = null;
session.lastCompletedSegment = null;
session.paused = false;
this.logger.debug(`Stopped transcoding for session ${session.id}`);
}
private pauseTranscode(session: Session) {
if (session.paused || !session.process) {
return;
}
session.process.kill('SIGSTOP');
session.paused = true;
this.logger.debug(`Paused transcoding for session ${session.id}`);
}
private resumeTranscode(session: Session) {
if (!session.paused || !session.process) {
return;
}
session.process.kill('SIGCONT');
session.paused = false;
this.logger.debug(`Resumed transcoding for session ${session.id}`);
}
private async removeSessionDir(session: { ownerId: string; id: string }) {
const dir = StorageCore.getHlsSessionFolder({ ownerId: session.ownerId, sessionId: session.id });
try {
await this.storageRepository.unlinkDir(dir, { recursive: true, force: true });
} catch (error) {
if ((error as NodeJS.ErrnoException)?.code !== 'ENOENT') {
throw error;
}
this.logger.warn(`Session dir ${dir} does not exist.`);
}
}
private removeInactiveSessions() {
const cutoff = Date.now() - HLS_INACTIVITY_TIMEOUT_MS;
const inactiveSessions = [...this.sessions.values()].filter((s) => s.lastActivityTime.getTime() < cutoff);
return Promise.all(
inactiveSessions.map(async (session) => {
try {
this.websocketRepository.serverSend('HlsSessionEnd', { sessionId: session.id });
await this.onSessionEnd({ sessionId: session.id });
} catch (error) {
this.logger.error(`Failed to sweep inactive HLS session ${session.id}: ${error}`);
}
}),
);
}
private removeExpiredSessions() {
return this.databaseRepository.withLock(DatabaseLock.HlsSessionCleanup, async () => {
const expiredSessions = await this.videoStreamRepository.getExpiredSessions();
await Promise.all(
expiredSessions.map(async (session) => {
await this.removeSessionDir(session);
await this.videoStreamRepository.deleteSession(session.id);
}),
);
});
}
}
+40 -12
View File
@@ -2,6 +2,7 @@ import { DateTime } from 'luxon';
import { SemVer } from 'semver';
import { defaults } from 'src/config';
import { serverVersion } from 'src/constants';
import { ReleaseChannel } from 'src/dtos/system-config.dto';
import { CronJob, JobName, JobStatus, SystemMetadataKey } from 'src/enum';
import { VersionService } from 'src/services/version.service';
import { factory } from 'test/small.factory';
@@ -22,6 +23,17 @@ describe(VersionService.name, () => {
mocks.cron.update.mockResolvedValue();
});
beforeAll(() => {
vitest.mock(import('src/constants.js'), async () => ({
...(await vitest.importActual<typeof import('src/constants.js')>('src/constants.js')),
serverVersion: new SemVer('v3.0.0'),
}));
});
afterAll(() => {
vitest.unmock(import('src/constants.js'));
});
it('should work', () => {
expect(sut).toBeDefined();
});
@@ -66,9 +78,10 @@ describe(VersionService.name, () => {
describe('getVersion', () => {
it('should respond the server version', () => {
expect(sut.getVersion()).toEqual({
major: serverVersion.major,
minor: serverVersion.minor,
patch: serverVersion.patch,
major: 3,
minor: 0,
patch: 0,
prerelease: null,
});
});
});
@@ -143,24 +156,24 @@ describe(VersionService.name, () => {
describe('onConfigUpdate', () => {
it('should queue a version check job when newVersionCheck is enabled', async () => {
await sut.onConfigUpdate({
oldConfig: { ...defaults, newVersionCheck: { enabled: false } },
newConfig: { ...defaults, newVersionCheck: { enabled: true } },
oldConfig: { ...defaults, newVersionCheck: { enabled: false, channel: ReleaseChannel.Stable } },
newConfig: { ...defaults, newVersionCheck: { enabled: true, channel: ReleaseChannel.Stable } },
});
expect(mocks.job.queue).toHaveBeenCalledWith({ name: JobName.VersionCheck, data: {} });
});
it('should not queue a version check job when newVersionCheck is disabled', async () => {
await sut.onConfigUpdate({
oldConfig: { ...defaults, newVersionCheck: { enabled: true } },
newConfig: { ...defaults, newVersionCheck: { enabled: false } },
oldConfig: { ...defaults, newVersionCheck: { enabled: true, channel: ReleaseChannel.Stable } },
newConfig: { ...defaults, newVersionCheck: { enabled: false, channel: ReleaseChannel.Stable } },
});
expect(mocks.job.queue).not.toHaveBeenCalled();
});
it('should not queue a version check job when newVersionCheck was already enabled', async () => {
await sut.onConfigUpdate({
oldConfig: { ...defaults, newVersionCheck: { enabled: true } },
newConfig: { ...defaults, newVersionCheck: { enabled: true } },
oldConfig: { ...defaults, newVersionCheck: { enabled: true, channel: ReleaseChannel.Stable } },
newConfig: { ...defaults, newVersionCheck: { enabled: true, channel: ReleaseChannel.Stable } },
});
expect(mocks.job.queue).not.toHaveBeenCalled();
});
@@ -169,21 +182,36 @@ describe(VersionService.name, () => {
describe('onWebsocketConnection', () => {
it('should send on_server_version client event', async () => {
await sut.onWebsocketConnection({ userId: '42' });
expect(mocks.websocket.clientSend).toHaveBeenCalledWith('on_server_version', '42', expect.any(SemVer));
expect(mocks.websocket.clientSend).toHaveBeenCalledWith('on_server_version', '42', {
major: 3,
minor: 0,
patch: 0,
prerelease: null,
});
expect(mocks.websocket.clientSend).toHaveBeenCalledTimes(1);
});
it('should also send a new release notification', async () => {
mocks.systemMetadata.get.mockResolvedValue({ checkedAt: '2024-01-01', releaseVersion: 'v1.42.0' });
await sut.onWebsocketConnection({ userId: '42' });
expect(mocks.websocket.clientSend).toHaveBeenCalledWith('on_server_version', '42', expect.any(SemVer));
expect(mocks.websocket.clientSend).toHaveBeenCalledWith('on_server_version', '42', {
major: 3,
minor: 0,
patch: 0,
prerelease: null,
});
expect(mocks.websocket.clientSend).toHaveBeenCalledWith('on_new_release', '42', expect.any(Object));
});
it('should not send a release notification when the version check is disabled', async () => {
mocks.systemMetadata.get.mockResolvedValueOnce({ newVersionCheck: { enabled: false } });
await sut.onWebsocketConnection({ userId: '42' });
expect(mocks.websocket.clientSend).toHaveBeenCalledWith('on_server_version', '42', expect.any(SemVer));
expect(mocks.websocket.clientSend).toHaveBeenCalledWith('on_server_version', '42', {
major: 3,
minor: 0,
patch: 0,
prerelease: null,
});
expect(mocks.websocket.clientSend).not.toHaveBeenCalledWith('on_new_release', '42', expect.any(Object));
});
});
+27 -8
View File
@@ -3,19 +3,27 @@ import { DateTime } from 'luxon';
import semver, { SemVer } from 'semver';
import { serverVersion } from 'src/constants';
import { OnEvent, OnJob } from 'src/decorators';
import { ReleaseNotification, ServerVersionResponseDto } from 'src/dtos/server.dto';
import { ReleaseEventV1, ReleaseType, ServerVersionResponseDto } from 'src/dtos/server.dto';
import { ReleaseChannel } from 'src/dtos/system-config.dto';
import { CronJob, DatabaseLock, ImmichWorker, JobName, JobStatus, QueueName, SystemMetadataKey } from 'src/enum';
import { ArgOf } from 'src/repositories/event.repository';
import { BaseService } from 'src/services/base.service';
import { VersionCheckMetadata } from 'src/types';
import { handlePromiseError } from 'src/utils/misc';
const asNotification = ({ checkedAt, releaseVersion }: VersionCheckMetadata): ReleaseNotification => {
const asNotification = (
channel: ReleaseChannel,
{ checkedAt, releaseVersion }: VersionCheckMetadata,
): ReleaseEventV1 => {
return {
isAvailable: semver.gt(releaseVersion, serverVersion),
// can't use gt because it's broken for release candidates F https://github.com/npm/node-semver/issues/483
isAvailable: semver.intersects(`>${serverVersion}`, releaseVersion.toString(), {
includePrerelease: channel === ReleaseChannel.ReleaseCandidate,
}),
checkedAt,
serverVersion: ServerVersionResponseDto.fromSemVer(serverVersion),
releaseVersion: ServerVersionResponseDto.fromSemVer(new SemVer(releaseVersion)),
type: semver.diff(serverVersion, releaseVersion) as ReleaseType,
};
};
@@ -98,14 +106,21 @@ export class VersionService extends BaseService {
}
}
const { version: releaseVersion, published_at: publishedAt } = await this.serverInfoRepository.getLatestRelease();
const { version: releaseVersion, published_at: publishedAt } = await this.serverInfoRepository.getLatestRelease(
newVersionCheck.channel,
);
const metadata: VersionCheckMetadata = { checkedAt: DateTime.utc().toISO(), releaseVersion };
await this.systemMetadataRepository.set(SystemMetadataKey.VersionCheckState, metadata);
if (semver.gt(releaseVersion, serverVersion)) {
// can't use gt because it's broken for release candidates F https://github.com/npm/node-semver/issues/483
if (
semver.intersects(`>${serverVersion}`, releaseVersion.toString(), {
includePrerelease: newVersionCheck.channel === ReleaseChannel.ReleaseCandidate,
})
) {
this.logger.log(`Found ${releaseVersion}, released at ${new Date(publishedAt).toLocaleString()}`);
this.websocketRepository.clientBroadcast('on_new_release', asNotification(metadata));
this.websocketRepository.clientBroadcast('on_new_release', asNotification(newVersionCheck.channel, metadata));
}
} catch (error: Error | any) {
this.logger.warn(`Unable to run version check: ${error}\n${error?.stack}`);
@@ -117,7 +132,11 @@ export class VersionService extends BaseService {
@OnEvent({ name: 'WebsocketConnect' })
async onWebsocketConnection({ userId }: ArgOf<'WebsocketConnect'>) {
this.websocketRepository.clientSend('on_server_version', userId, serverVersion);
this.websocketRepository.clientSend(
'on_server_version',
userId,
ServerVersionResponseDto.fromSemVer(serverVersion),
);
const { newVersionCheck } = await this.getConfig({ withCache: true });
if (!newVersionCheck.enabled) {
@@ -126,7 +145,7 @@ export class VersionService extends BaseService {
const metadata = await this.systemMetadataRepository.get(SystemMetadataKey.VersionCheckState);
if (metadata) {
this.websocketRepository.clientSend('on_new_release', userId, asNotification(metadata));
this.websocketRepository.clientSend('on_new_release', userId, asNotification(newVersionCheck.channel, metadata));
}
}
}
+103 -32
View File
@@ -1,10 +1,15 @@
import { CurrentPlugin } from '@extism/extism';
import { WorkflowChanges, WorkflowEventData, WorkflowEventPayload, WorkflowResponse } from '@immich/plugin-sdk';
import {
WorkflowChanges,
WorkflowEventData,
WorkflowEventPayload,
WorkflowResponse,
WorkflowTrigger,
} from '@immich/plugin-sdk';
import { HttpException, UnauthorizedException } from '@nestjs/common';
import _ from 'lodash';
import { join } from 'node:path';
import { OnEvent, OnJob } from 'src/decorators';
import { AlbumsAddAssetsDto } from 'src/dtos/album.dto';
import { DummyValue, OnEvent, OnJob } from 'src/decorators';
import { AlbumsAddAssetsDto, CreateAlbumDto, GetAlbumsDto } from 'src/dtos/album.dto';
import { BulkIdsDto } from 'src/dtos/asset-ids.response.dto';
import { AuthDto } from 'src/dtos/auth.dto';
import { PluginManifestDto } from 'src/dtos/plugin-manifest.dto';
@@ -16,11 +21,11 @@ import {
JobName,
JobStatus,
QueueName,
WorkflowTrigger,
WorkflowType,
} from 'src/enum';
import { ArgOf } from 'src/repositories/event.repository';
import { AlbumService } from 'src/services/album.service';
import { AssetService } from 'src/services/asset.service';
import { BaseService } from 'src/services/base.service';
import { JobOf } from 'src/types';
@@ -32,9 +37,11 @@ const dummy = () => {
type ExecuteOptions<T extends WorkflowType> = {
read: (type: T) => Promise<{ authUserId: string; data: WorkflowEventData<T> }>;
write: (changes: WorkflowChanges<T>) => Promise<void>;
write: (auth: AuthDto, changes: WorkflowChanges<T>) => Promise<void>;
};
type AssetTrigger = { userId: string; assetId: string; trigger: WorkflowTrigger };
export class WorkflowExecutionService extends BaseService {
private jwtSecret!: string;
@@ -59,21 +66,26 @@ export class WorkflowExecutionService extends BaseService {
const albumService = BaseService.create(AlbumService, this);
const albumAddAssets = this.wrap<[id: string, dto: BulkIdsDto]>((authDto, args) =>
const searchAlbums = this.wrap<[dto: GetAlbumsDto]>((authDto, args) => albumService.getAll(authDto, ...args));
const createAlbum = this.wrap<[dto: CreateAlbumDto]>((authDto, args) => albumService.create(authDto, ...args));
const addAssetsToAlbum = this.wrap<[id: string, dto: BulkIdsDto]>((authDto, args) =>
albumService.addAssets(authDto, ...args),
);
const addAssetsToAlbums = this.wrap<[dto: AlbumsAddAssetsDto]>((authDto, args) =>
albumService.addAssetsToAlbums(authDto, ...args),
);
const functions = {
albumAddAssets,
searchAlbums,
createAlbum,
addAssetsToAlbum,
addAssetsToAlbums,
};
const stubs = {
albumAddAssets: dummy,
const stubs: typeof functions = {
searchAlbums: dummy,
createAlbum: dummy,
addAssetsToAlbum: dummy,
addAssetsToAlbums: dummy,
};
@@ -247,20 +259,36 @@ export class WorkflowExecutionService extends BaseService {
}
@OnEvent({ name: 'AssetCreate' })
async onAssetCreate({ asset }: ArgOf<'AssetCreate'>) {
const dto = { ownerId: asset.ownerId, trigger: WorkflowTrigger.AssetCreate };
const items = await this.workflowRepository.search(dto);
onAssetCreate({ asset: { ownerId: userId, id: assetId } }: ArgOf<'AssetCreate'>) {
return this.onAssetTrigger({ userId, assetId, trigger: WorkflowTrigger.AssetCreate });
}
@OnEvent({ name: 'AssetMetadataExtracted' })
onAssetMetadataExtracted({ userId, assetId, source }: ArgOf<'AssetMetadataExtracted'>) {
// prevent loops
// TODO loop detection in job service directly
if (source === 'sidecar-write') {
return;
}
return this.onAssetTrigger({ userId, assetId, trigger: WorkflowTrigger.AssetMetadataExtraction });
}
private async onAssetTrigger({ userId, assetId, trigger }: AssetTrigger) {
const items = await this.workflowRepository.search({ userId, trigger });
await this.jobRepository.queueAll(
items.map((workflow) => ({
name: JobName.WorkflowAssetCreate,
data: { workflowId: workflow.id, assetId: asset.id },
name: JobName.WorkflowAssetTrigger,
data: { workflowId: workflow.id, assetId, trigger },
})),
);
}
@OnJob({ name: JobName.WorkflowAssetCreate, queue: QueueName.Workflow })
handleAssetCreate({ workflowId, assetId }: JobOf<JobName.WorkflowAssetCreate>) {
@OnJob({ name: JobName.WorkflowAssetTrigger, queue: QueueName.Workflow })
handleAssetTrigger({ workflowId, assetId }: JobOf<JobName.WorkflowAssetTrigger>) {
return this.execute(workflowId, (type) => {
const assetService = BaseService.create(AssetService, this);
switch (type) {
case WorkflowType.AssetV1: {
return {
@@ -271,19 +299,35 @@ export class WorkflowExecutionService extends BaseService {
authUserId: asset.ownerId,
};
},
write: async (changes) => {
if (changes.asset) {
await this.assetRepository.update({
id: assetId,
..._.omitBy(
{
isFavorite: changes.asset?.isFavorite,
visibility: changes.asset?.visibility,
},
_.isUndefined,
),
});
write: async (auth, changes) => {
const asset = changes.asset;
if (!asset) {
return;
}
await assetService.update(auth, assetId, {
isFavorite: asset.isFavorite,
visibility: asset.visibility,
dateTimeOriginal: asset.exifInfo?.dateTimeOriginal ?? undefined,
// TODO allow setting to null
longitude: asset.exifInfo?.longitude ?? undefined,
// TODO allow setting to null
latitude: asset.exifInfo?.latitude ?? undefined,
// TODO allow setting to null
description: asset.exifInfo?.description ?? undefined,
rating: asset.exifInfo?.rating,
// TODO add to update dto
// make: asset.exifInfo?.make,
// model: asset.exifInfo?.model,
// city: asset.exifInfo?.city,
// state: asset.exifInfo?.state,
// country: asset.exifInfo?.country,
// lensModel: asset.exifInfo?.lensModel,
// fNumber: asset.exifInfo?.fNumber,
// fps: asset.exifInfo?.fps,
// iso: asset.exifInfo?.iso,
});
},
} satisfies ExecuteOptions<typeof type>;
}
@@ -301,7 +345,19 @@ export class WorkflowExecutionService extends BaseService {
}
// TODO infer from steps
const type = 'AssetV1' as T;
let type: T | undefined;
for (const targetType of Object.values(WorkflowType)) {
const missing = workflow.steps.some((step) => !step.types.includes(targetType));
if (!missing) {
type = targetType as unknown as T;
break;
}
}
if (!type) {
throw new Error('Unable to infer workflow event type from steps');
}
const handler = getHandler(type);
if (!handler) {
this.logger.error(`Misconfigured workflow ${workflowId}: no handler for type ${type}`);
@@ -337,10 +393,25 @@ export class WorkflowExecutionService extends BaseService {
payload,
);
if (result?.changes) {
await write(result.changes);
await write(
{
user: {
id: readResult.authUserId,
},
session: {
id: DummyValue.UUID,
hasElevatedPermission: true,
},
} as AuthDto,
result.changes,
);
({ data } = await read(type));
}
if (result?.config) {
await this.workflowRepository.updateStep(step.id, { config: result.config });
}
const shouldContinue = result?.workflow?.continue ?? true;
if (!shouldContinue) {
break;
+3 -3
View File
@@ -1,4 +1,4 @@
import { WorkflowStepConfig } from '@immich/plugin-sdk';
import { WorkflowStepConfig, WorkflowTrigger } from '@immich/plugin-sdk';
import { BadRequestException, Injectable } from '@nestjs/common';
import { AuthDto } from 'src/dtos/auth.dto';
import {
@@ -11,7 +11,7 @@ import {
WorkflowTriggerResponseDto,
WorkflowUpdateDto,
} from 'src/dtos/workflow.dto';
import { Permission, WorkflowTrigger } from 'src/enum';
import { Permission } from 'src/enum';
import { PluginMethodSearchResponse } from 'src/repositories/plugin.repository';
import { BaseService } from 'src/services/base.service';
import { getWorkflowTriggers, isMethodCompatible, resolveMethod } from 'src/utils/workflow';
@@ -23,7 +23,7 @@ export class WorkflowService extends BaseService {
}
async search(auth: AuthDto, dto: WorkflowSearchDto): Promise<WorkflowResponseDto[]> {
const workflows = await this.workflowRepository.search({ ...dto, ownerId: auth.user.id });
const workflows = await this.workflowRepository.search({ ...dto, userId: auth.user.id });
return workflows.map((workflow) => mapWorkflow(workflow));
}
+26 -9
View File
@@ -1,3 +1,4 @@
import { WorkflowTrigger } from '@immich/plugin-sdk';
import { ShallowDehydrateObject } from 'kysely';
import { SystemConfig } from 'src/config';
import { VECTOR_EXTENSIONS } from 'src/constants';
@@ -28,8 +29,6 @@ import {
SystemMetadataKey,
TranscodeTarget,
UserMetadataKey,
VideoCodec,
WorkflowTrigger,
WorkflowType,
} from 'src/enum';
@@ -162,6 +161,25 @@ export interface TranscodeCommand {
};
}
export interface VideoTuning {
strictGop: boolean;
lowLatency: boolean;
}
export interface HlsCommandOptions {
initFilename: string;
inputPath: string;
packetCount: number;
playlistFilename: string;
seekSeconds?: number;
segmentDuration: number;
segmentFilename: string;
startSegment: number;
target: TranscodeTarget;
timeBase: number;
totalDuration: number;
}
export interface BitrateDistribution {
max: number;
target: number;
@@ -177,14 +195,11 @@ export interface ImageBuffer {
export interface VideoCodecSWConfig {
getCommand(
target: TranscodeTarget,
videoStream: VideoStreamInfo,
audioStream?: AudioStreamInfo,
video: VideoStreamInfo,
audio?: AudioStreamInfo,
format?: VideoFormat,
): TranscodeCommand;
}
export interface VideoCodecHWConfig extends VideoCodecSWConfig {
getSupportedCodecs(): Array<VideoCodec>;
getHlsCommand(options: HlsCommandOptions, video: VideoStreamInfo, audio?: AudioStreamInfo): string[];
}
export interface ProbeOptions {
@@ -371,6 +386,7 @@ export type JobItem =
// Cleanup
| { name: JobName.SessionCleanup; data?: IBaseJob }
| { name: JobName.HlsSessionCleanup; data?: IBaseJob }
// Tags
| { name: JobName.TagCleanup; data?: IBaseJob }
@@ -404,7 +420,7 @@ export type JobItem =
| { name: JobName.Ocr; data: IEntityJob }
// Workflow
| { name: JobName.WorkflowAssetCreate; data: { workflowId: string; assetId: string } }
| { name: JobName.WorkflowAssetTrigger; data: { workflowId: string; assetId: string } }
// Editor
| { name: JobName.AssetEditThumbnailGeneration; data: IEntityJob };
@@ -523,6 +539,7 @@ export type UserPreferences = {
people: {
enabled: boolean;
sidebarWeb: boolean;
minimumFaces: number;
};
ratings: {
enabled: boolean;
+6 -3
View File
@@ -71,10 +71,13 @@ export const removeUndefinedKeys = <T extends object>(update: T, template: unkno
};
export const ASSET_CHECKSUM_CONSTRAINT = 'UQ_assets_owner_checksum';
export const VIDEO_STREAM_SESSION_PK_CONSTRAINT = 'video_stream_session_pkey';
export const isAssetChecksumConstraint = (error: unknown) => {
return (error as PostgresError)?.constraint_name === 'UQ_assets_owner_checksum';
};
export const isAssetChecksumConstraint = (error: unknown) =>
(error as PostgresError)?.constraint_name === ASSET_CHECKSUM_CONSTRAINT;
export const isVideoStreamSessionPkConstraint = (error: unknown) =>
(error as PostgresError)?.constraint_name === VIDEO_STREAM_SESSION_PK_CONSTRAINT;
export function withDefaultVisibility<O>(qb: SelectQueryBuilder<DB, 'asset', O>) {
return qb.where('asset.visibility', 'in', [sql.lit(AssetVisibility.Archive), sql.lit(AssetVisibility.Timeline)]);
+50
View File
@@ -0,0 +1,50 @@
import { ArgOf, EmitEvent } from 'src/repositories/event.repository';
export class PendingEvents<T extends { [T in EmitEvent]: ArgOf<T> extends { error?: string } ? T : never }[EmitEvent]> {
private pending = new Map<string, { completers: PromiseWithResolvers<ArgOf<T>>[]; timeout: NodeJS.Timeout }>();
private timeoutMs: number;
constructor({ timeoutMs }: { timeoutMs: number }) {
this.timeoutMs = timeoutMs;
}
wait(key: string): Promise<ArgOf<T>> {
const completer = Promise.withResolvers<ArgOf<T>>();
const existing = this.pending.get(key);
if (existing) {
existing.completers.push(completer);
return completer.promise;
}
const timeout = setTimeout(() => this.complete(key, { error: 'Request timed out' }), this.timeoutMs);
this.pending.set(key, { completers: [completer], timeout });
return completer.promise;
}
complete(key: string, value: ArgOf<T> | { error: string }) {
const pending = this.pending.get(key);
if (!pending) {
return;
}
clearTimeout(pending.timeout);
this.pending.delete(key);
if ('error' in value) {
const error = new Error(value.error);
for (const completer of pending.completers) {
completer.reject(error);
}
} else {
for (const completer of pending.completers) {
completer.resolve(value);
}
}
}
rejectByPrefix(prefix: string, error: string) {
for (const key of this.pending.keys()) {
if (key.startsWith(prefix)) {
this.complete(key, { error });
}
}
}
}
+184 -136
View File
@@ -1,4 +1,4 @@
import { AUDIO_ENCODER } from 'src/constants';
import { AUDIO_ENCODER, SUPPORTED_HWA_CODECS } from 'src/constants';
import { SystemConfigFFmpegDto } from 'src/dtos/system-config.dto';
import {
ColorMatrix,
@@ -13,38 +13,56 @@ import {
import {
AudioStreamInfo,
BitrateDistribution,
HlsCommandOptions,
TranscodeCommand,
VideoCodecHWConfig,
VideoCodecSWConfig,
VideoFormat,
VideoInterfaces,
VideoStreamInfo,
VideoTuning,
} from 'src/types';
export const isVideoRotated = (videoStream: VideoStreamInfo): boolean => Math.abs(videoStream.rotation) === 90;
export const isVideoVertical = (videoStream: VideoStreamInfo): boolean =>
videoStream.height > videoStream.width || isVideoRotated(videoStream);
export const getOutputSize = (videoStream: VideoStreamInfo, targetRes: number) => {
const factor = Math.max(videoStream.height, videoStream.width) / Math.min(videoStream.height, videoStream.width);
let larger = Math.round(targetRes * factor);
if (larger % 2 !== 0) {
larger -= 1;
}
return isVideoVertical(videoStream) ? { width: targetRes, height: larger } : { width: larger, height: targetRes };
};
export class BaseConfig implements VideoCodecSWConfig {
readonly presets = ['veryslow', 'slower', 'slow', 'medium', 'fast', 'faster', 'veryfast', 'superfast', 'ultrafast'];
protected constructor(protected config: SystemConfigFFmpegDto) {}
protected constructor(
protected config: SystemConfigFFmpegDto,
protected tune: VideoTuning = { strictGop: false, lowLatency: false },
) {}
static create(config: SystemConfigFFmpegDto, interfaces: VideoInterfaces): VideoCodecSWConfig {
static create(config: SystemConfigFFmpegDto, interfaces: VideoInterfaces, tune?: VideoTuning) {
if (config.accel === TranscodeHardwareAcceleration.Disabled) {
return this.getSWCodecConfig(config);
return this.getSWCodecConfig(config, tune);
}
return this.getHWCodecConfig(config, interfaces);
return this.getHWCodecConfig(config, interfaces, tune);
}
private static getSWCodecConfig(config: SystemConfigFFmpegDto) {
private static getSWCodecConfig(config: SystemConfigFFmpegDto, tune?: VideoTuning): VideoCodecSWConfig {
switch (config.targetVideoCodec) {
case VideoCodec.H264: {
return new H264Config(config);
return new H264Config(config, tune);
}
case VideoCodec.Hevc: {
return new HEVCConfig(config);
return new HEVCConfig(config, tune);
}
case VideoCodec.Vp9: {
return new VP9Config(config);
return new VP9Config(config, tune);
}
case VideoCodec.Av1: {
return new AV1Config(config);
return new AV1Config(config, tune);
}
default: {
throw new Error(`Codec '${config.targetVideoCodec}' is unsupported`);
@@ -52,72 +70,122 @@ export class BaseConfig implements VideoCodecSWConfig {
}
}
private static getHWCodecConfig(config: SystemConfigFFmpegDto, interfaces: VideoInterfaces) {
let handler: VideoCodecHWConfig;
private static getHWCodecConfig(config: SystemConfigFFmpegDto, interfaces: VideoInterfaces, tune?: VideoTuning) {
if (!SUPPORTED_HWA_CODECS[config.accel].includes(config.targetVideoCodec)) {
throw new Error(
`${config.accel.toUpperCase()} acceleration does not support codec '${config.targetVideoCodec.toUpperCase()}'. Supported codecs: ${SUPPORTED_HWA_CODECS[config.accel]}`,
);
}
let handler: VideoCodecSWConfig;
switch (config.accel) {
case TranscodeHardwareAcceleration.Nvenc: {
handler = config.accelDecode
? new NvencHwDecodeConfig(config, interfaces)
: new NvencSwDecodeConfig(config, interfaces);
? new NvencHwDecodeConfig(config, interfaces, tune)
: new NvencSwDecodeConfig(config, interfaces, tune);
break;
}
case TranscodeHardwareAcceleration.Qsv: {
handler = config.accelDecode
? new QsvHwDecodeConfig(config, interfaces)
: new QsvSwDecodeConfig(config, interfaces);
? new QsvHwDecodeConfig(config, interfaces, tune)
: new QsvSwDecodeConfig(config, interfaces, tune);
break;
}
case TranscodeHardwareAcceleration.Vaapi: {
handler = config.accelDecode
? new VaapiHwDecodeConfig(config, interfaces)
: new VaapiSwDecodeConfig(config, interfaces);
? new VaapiHwDecodeConfig(config, interfaces, tune)
: new VaapiSwDecodeConfig(config, interfaces, tune);
break;
}
case TranscodeHardwareAcceleration.Rkmpp: {
handler = config.accelDecode
? new RkmppHwDecodeConfig(config, interfaces)
: new RkmppSwDecodeConfig(config, interfaces);
? new RkmppHwDecodeConfig(config, interfaces, tune)
: new RkmppSwDecodeConfig(config, interfaces, tune);
break;
}
default: {
throw new Error(`${config.accel.toUpperCase()} acceleration is unsupported`);
}
}
if (!handler.getSupportedCodecs().includes(config.targetVideoCodec)) {
throw new Error(
`${config.accel.toUpperCase()} acceleration does not support codec '${config.targetVideoCodec.toUpperCase()}'. Supported codecs: ${handler.getSupportedCodecs()}`,
);
}
return handler;
}
getCommand(
target: TranscodeTarget,
videoStream: VideoStreamInfo,
audioStream?: AudioStreamInfo,
format?: VideoFormat,
) {
getCommand(target: TranscodeTarget, video: VideoStreamInfo, audio?: AudioStreamInfo, format?: VideoFormat) {
const options = {
inputOptions: this.getBaseInputOptions(videoStream, format),
outputOptions: [...this.getBaseOutputOptions(target, videoStream, audioStream), '-v', 'verbose'],
inputOptions: this.getBaseInputOptions(video, format),
outputOptions: [
...this.getBaseOutputOptions(target, video, audio),
...this.getPresetOptions(),
...this.getBitrateOptions(),
...this.getEncoderOptions(),
'-movflags',
'faststart',
'-fps_mode',
'passthrough',
'-v',
'verbose',
],
twoPass: this.eligibleForTwoPass(),
progress: { frameCount: videoStream.frameCount, percentInterval: 5 },
progress: { frameCount: video.frameCount, percentInterval: 5 },
} as TranscodeCommand;
if ([TranscodeTarget.All, TranscodeTarget.Video].includes(target)) {
const filters = this.getFilterOptions(videoStream);
const filters = this.getFilterOptions(video);
if (filters.length > 0) {
options.outputOptions.push('-vf', filters.join(','));
}
}
options.outputOptions.push(
return options;
}
getHlsCommand(options: HlsCommandOptions, video: VideoStreamInfo, audio?: AudioStreamInfo) {
const args: string[] = this.getBaseInputOptions(video);
if (options.seekSeconds) {
args.push('-ss', String(options.seekSeconds));
}
args.push(
'-nostdin',
'-nostats',
'-i',
options.inputPath,
...this.getBaseOutputOptions(options.target, video, audio),
...this.getPresetOptions(),
...this.getOutputThreadOptions(),
...this.getBitrateOptions(),
...this.getEncoderOptions(),
'-copyts',
'-r',
`${options.packetCount * options.timeBase}/${options.totalDuration}`,
'-avoid_negative_ts',
'disabled',
'-f',
'hls',
'-hls_time',
String(options.segmentDuration),
'-hls_list_size',
'0',
'-hls_segment_type',
'fmp4',
'-hls_fmp4_init_filename',
options.initFilename,
'-hls_segment_options',
'movflags=+frag_discont',
'-hls_flags',
'temp_file',
'-hls_segment_filename',
options.segmentFilename,
'-start_number',
String(options.startSegment),
);
return options;
if ([TranscodeTarget.All, TranscodeTarget.Video].includes(options.target)) {
const filters = this.getFilterOptions(video);
if (filters.length > 0) {
args.push('-vf', filters.join(','));
}
}
args.push(options.playlistFilename);
return args;
}
// eslint-disable-next-line @typescript-eslint/no-unused-vars
@@ -129,23 +197,7 @@ export class BaseConfig implements VideoCodecSWConfig {
const videoCodec = [TranscodeTarget.All, TranscodeTarget.Video].includes(target) ? this.getVideoCodec() : 'copy';
const audioCodec = [TranscodeTarget.All, TranscodeTarget.Audio].includes(target) ? this.getAudioEncoder() : 'copy';
const options = [
'-c:v',
videoCodec,
'-c:a',
audioCodec,
// Makes a second pass moving the moov atom to the
// beginning of the file for improved playback speed.
'-movflags',
'faststart',
'-fps_mode',
'passthrough',
'-map',
`0:${videoStream.index}`,
'-map_metadata',
'-1',
];
const options = ['-c:v', videoCodec, '-c:a', audioCodec, '-map', `0:${videoStream.index}`, '-map_metadata', '-1'];
if (audioStream) {
options.push('-map', `0:${audioStream.index}`);
}
@@ -157,18 +209,22 @@ export class BaseConfig implements VideoCodecSWConfig {
}
if (this.getGopSize() > 0) {
options.push('-g', `${this.getGopSize()}`);
if (this.tune.strictGop) {
options.push('-keyint_min', `${this.getGopSize()}`);
}
}
if (
this.config.targetVideoCodec === VideoCodec.Hevc &&
(videoCodec !== 'copy' || videoStream.codecName === 'hevc')
) {
const isHvc = (videoCodec === 'copy' ? videoStream.codecName : videoCodec) === VideoCodec.Hevc;
if (isHvc) {
options.push('-tag:v', 'hvc1');
}
return options;
}
getEncoderOptions(): string[] {
return [];
}
getFilterOptions(videoStream: VideoStreamInfo) {
const options = [];
if (this.shouldScale(videoStream)) {
@@ -272,25 +328,7 @@ export class BaseConfig implements VideoCodecSWConfig {
getScaling(videoStream: VideoStreamInfo, mult = 2) {
const targetResolution = this.getTargetResolution(videoStream);
return this.isVideoVertical(videoStream) ? `${targetResolution}:-${mult}` : `-${mult}:${targetResolution}`;
}
getSize(videoStream: VideoStreamInfo) {
const smaller = this.getTargetResolution(videoStream);
const factor = Math.max(videoStream.height, videoStream.width) / Math.min(videoStream.height, videoStream.width);
let larger = Math.round(smaller * factor);
if (larger % 2 !== 0) {
larger -= 1;
}
return this.isVideoVertical(videoStream) ? { width: smaller, height: larger } : { width: larger, height: smaller };
}
isVideoRotated(videoStream: VideoStreamInfo) {
return Math.abs(videoStream.rotation) === 90;
}
isVideoVertical(videoStream: VideoStreamInfo) {
return videoStream.height > videoStream.width || this.isVideoRotated(videoStream);
return isVideoVertical(videoStream) ? `${targetResolution}:-${mult}` : `-${mult}:${targetResolution}`;
}
isBitrateConstrained() {
@@ -353,23 +391,18 @@ export class BaseConfig implements VideoCodecSWConfig {
}
}
export class BaseHWConfig extends BaseConfig implements VideoCodecHWConfig {
export class BaseHWConfig extends BaseConfig {
protected device: string;
protected interfaces: VideoInterfaces;
constructor(
protected config: SystemConfigFFmpegDto,
interfaces: VideoInterfaces,
protected interfaces: VideoInterfaces,
tune?: VideoTuning,
) {
super(config);
this.interfaces = interfaces;
super(config, tune);
this.device = this.getDevice(interfaces);
}
getSupportedCodecs() {
return [VideoCodec.H264, VideoCodec.Hevc];
}
validateDevices(devices: string[]) {
if (devices.length === 0) {
throw new Error('No /dev/dri devices found. If using Docker, make sure at least one /dev/dri device is mounted');
@@ -474,24 +507,32 @@ export class ThumbnailConfig extends BaseConfig {
}
export class H264Config extends BaseConfig {
getOutputThreadOptions() {
const options = super.getOutputThreadOptions();
if (this.config.threads === 1) {
options.push('-x264-params', 'frame-threads=1:pools=none');
getEncoderOptions(): string[] {
const out = this.getOutputThreadOptions();
if (this.tune.strictGop) {
out.push('-sc_threshold:v', '0');
}
return options;
if (this.config.threads === 1) {
out.push('-x264-params', 'frame-threads=1:pools=none');
}
return out;
}
}
export class HEVCConfig extends BaseConfig {
getOutputThreadOptions() {
const options = super.getOutputThreadOptions();
if (this.config.threads === 1) {
options.push('-x265-params', 'frame-threads=1:pools=none');
getEncoderOptions(): string[] {
const out: string[] = this.getOutputThreadOptions();
const params: string[] = [];
if (this.tune.strictGop) {
params.push('no-scenecut=1', 'no-open-gop=1');
}
return options;
if (this.config.threads === 1) {
params.push('frame-threads=1', 'pools=none');
}
if (params.length > 0) {
out.push('-x265-params', params.join(':'));
}
return out;
}
}
@@ -520,8 +561,8 @@ export class VP9Config extends BaseConfig {
return [`-${this.useCQP() ? 'q:v' : 'crf'}`, `${this.config.crf}`, '-b:v', `${bitrates.max}${bitrates.unit}`];
}
getOutputThreadOptions() {
return ['-row-mt', '1', ...super.getOutputThreadOptions()];
getEncoderOptions(): string[] {
return ['-row-mt', '1', ...this.getOutputThreadOptions()];
}
eligibleForTwoPass() {
@@ -543,23 +584,22 @@ export class AV1Config extends BaseConfig {
}
getBitrateOptions() {
const options = ['-crf', `${this.config.crf}`];
const bitrates = this.getBitrateDistribution();
const svtparams = [];
if (this.config.threads > 0) {
svtparams.push(`lp=${this.config.threads}`);
}
if (bitrates.max > 0) {
svtparams.push(`mbr=${bitrates.max}${bitrates.unit}`);
}
if (svtparams.length > 0) {
options.push('-svtav1-params', svtparams.join(':'));
}
return options;
return ['-crf', `${this.config.crf}`];
}
getOutputThreadOptions() {
return []; // Already set above with svtav1-params
getEncoderOptions(): string[] {
const params: string[] = [];
if (this.tune.lowLatency) {
params.push('hierarchical-levels=3', 'lookahead=0', 'enable-tf=0');
}
if (this.config.threads > 0) {
params.push(`lp=${this.config.threads}`);
}
const bitrates = this.getBitrateDistribution();
if (bitrates.max > 0) {
params.push(`mbr=${bitrates.max}${bitrates.unit}`);
}
return params.length > 0 ? ['-svtav1-params', params.join(':')] : [];
}
eligibleForTwoPass() {
@@ -572,10 +612,6 @@ export class NvencSwDecodeConfig extends BaseHWConfig {
return '0';
}
getSupportedCodecs() {
return [VideoCodec.H264, VideoCodec.Hevc, VideoCodec.Av1];
}
getBaseInputOptions() {
return ['-init_hw_device', `cuda=cuda:${this.device}`, '-filter_hw_device', 'cuda'];
}
@@ -652,6 +688,14 @@ export class NvencSwDecodeConfig extends BaseHWConfig {
return [];
}
getEncoderOptions(): string[] {
const out = this.getOutputThreadOptions();
if (this.tune.strictGop) {
out.push('-forced-idr', '1');
}
return out;
}
getRefs() {
const bframes = this.getBFrames();
if (bframes > 0 && bframes < 3 && this.config.refs < 3) {
@@ -703,8 +747,8 @@ export class NvencHwDecodeConfig extends NvencSwDecodeConfig {
return ['-threads', '1'];
}
getOutputThreadOptions() {
return [];
getEncoderOptions(): string[] {
return this.tune.strictGop ? ['-forced-idr', '1'] : [];
}
}
@@ -749,10 +793,6 @@ export class QsvSwDecodeConfig extends BaseHWConfig {
return options;
}
getSupportedCodecs() {
return [VideoCodec.H264, VideoCodec.Hevc, VideoCodec.Vp9, VideoCodec.Av1];
}
// recommended from https://github.com/intel/media-delivery/blob/master/doc/benchmarks/intel-iris-xe-max-graphics/intel-iris-xe-max-graphics.md
getBFrames() {
if (this.config.bframes < 0) {
@@ -775,6 +815,14 @@ export class QsvSwDecodeConfig extends BaseHWConfig {
getScaling(videoStream: VideoStreamInfo): string {
return super.getScaling(videoStream, 1);
}
getEncoderOptions(): string[] {
const out = this.getOutputThreadOptions();
if (this.tune.strictGop) {
out.push('-idr_interval', '0');
}
return out;
}
}
export class QsvHwDecodeConfig extends QsvSwDecodeConfig {
@@ -888,13 +936,17 @@ export class VaapiSwDecodeConfig extends BaseHWConfig {
return options;
}
getSupportedCodecs() {
return [VideoCodec.H264, VideoCodec.Hevc, VideoCodec.Vp9, VideoCodec.Av1];
}
useCQP() {
return this.config.cqMode !== CQMode.Icq || this.config.targetVideoCodec === VideoCodec.Vp9;
}
getEncoderOptions(): string[] {
const out = this.getOutputThreadOptions();
if (this.tune.strictGop) {
out.push('-idr_interval', '0');
}
return out;
}
}
export class VaapiHwDecodeConfig extends VaapiSwDecodeConfig {
@@ -988,10 +1040,6 @@ export class RkmppSwDecodeConfig extends BaseHWConfig {
return ['-rc_mode', 'CQP', '-qp_init', `${this.config.crf}`];
}
getSupportedCodecs() {
return [VideoCodec.H264, VideoCodec.Hevc];
}
getVideoCodec(): string {
return `${this.config.targetVideoCodec}_rkmpp`;
}
+2 -2
View File
@@ -15,7 +15,7 @@ import picomatch from 'picomatch';
import parse from 'picomatch/lib/parse';
import { SystemConfig } from 'src/config';
import { CLIP_MODEL_INFO, endpointTags, serverVersion } from 'src/constants';
import { extraSyncModels } from 'src/dtos/sync.dto';
import { extraModels } from 'src/decorators';
import { ApiCustomExtension, ImmichCookie, ImmichHeader, MetadataKey } from 'src/enum';
import { LoggingRepository } from 'src/repositories/logging.repository';
@@ -289,7 +289,7 @@ export const useSwagger = (app: INestApplication, { write }: { write: boolean })
const options: SwaggerDocumentOptions = {
operationIdFactory: (controllerKey: string, methodKey: string) => methodKey,
extraModels: extraSyncModels,
extraModels,
ignoreGlobalPrefix: true,
};
+1
View File
@@ -21,6 +21,7 @@ const getDefaultPreferences = (): UserPreferences => {
people: {
enabled: true,
sidebarWeb: false,
minimumFaces: 3,
},
sharedLinks: {
enabled: true,
+2 -1
View File
@@ -1,4 +1,5 @@
import { WorkflowTrigger, WorkflowType } from 'src/enum';
import { WorkflowTrigger } from '@immich/plugin-sdk';
import { WorkflowType } from 'src/enum';
import { isMethodCompatible } from 'src/utils/workflow';
const tests: Array<{ trigger: WorkflowTrigger; types: WorkflowType[]; expected: boolean }> = [
+3 -1
View File
@@ -1,9 +1,11 @@
import { WorkflowTrigger, WorkflowType } from 'src/enum';
import { WorkflowTrigger } from '@immich/plugin-sdk';
import { WorkflowType } from 'src/enum';
import { PluginMethodSearchResponse } from 'src/repositories/plugin.repository';
export const triggerMap: Record<WorkflowTrigger, WorkflowType[]> = {
[WorkflowTrigger.AssetCreate]: [WorkflowType.AssetV1],
[WorkflowTrigger.PersonRecognized]: [WorkflowType.AssetPersonV1],
[WorkflowTrigger.AssetMetadataExtraction]: [WorkflowType.AssetV1],
};
export const getWorkflowTriggers = () =>