Merge remote-tracking branch 'origin/lighter_buckets_web' into lighter_buckets_server

server/src/config.ts

@@ -5,6 +5,7 @@ import {
   CQMode,
   ImageFormat,
   LogLevel,
+  OAuthTokenEndpointAuthMethod,
   QueueName,
   ToneMapping,
   TranscodeHWAccel,
@@ -96,6 +97,8 @@ export interface SystemConfig {
     scope: string;
     signingAlgorithm: string;
     profileSigningAlgorithm: string;
+    tokenEndpointAuthMethod: OAuthTokenEndpointAuthMethod;
+    timeout: number;
     storageLabelClaim: string;
     storageQuotaClaim: string;
   };
@@ -260,6 +263,8 @@ export const defaults = Object.freeze<SystemConfig>({
     profileSigningAlgorithm: 'none',
     storageLabelClaim: 'preferred_username',
     storageQuotaClaim: 'immich_quota',
+    tokenEndpointAuthMethod: OAuthTokenEndpointAuthMethod.CLIENT_SECRET_POST,
+    timeout: 30_000,
   },
   passwordLogin: {
     enabled: true,

server/src/controllers/asset.controller.ts

@@ -1,7 +1,7 @@
 import { Body, Controller, Delete, Get, HttpCode, HttpStatus, Param, Post, Put, Query } from '@nestjs/common';
 import { ApiOperation, ApiTags } from '@nestjs/swagger';
 import { EndpointLifecycle } from 'src/decorators';
-import { AssetResponseDto, MemoryLaneResponseDto } from 'src/dtos/asset-response.dto';
+import { AssetResponseDto } from 'src/dtos/asset-response.dto';
 import {
   AssetBulkDeleteDto,
   AssetBulkUpdateDto,
@@ -13,7 +13,6 @@ import {
   UpdateAssetDto,
 } from 'src/dtos/asset.dto';
 import { AuthDto } from 'src/dtos/auth.dto';
-import { MemoryLaneDto } from 'src/dtos/search.dto';
 import { RouteKey } from 'src/enum';
 import { Auth, Authenticated } from 'src/middleware/auth.guard';
 import { AssetService } from 'src/services/asset.service';
@@ -24,12 +23,6 @@ import { UUIDParamDto } from 'src/validation';
 export class AssetController {
   constructor(private service: AssetService) {}
 
-  @Get('memory-lane')
-  @Authenticated()
-  getMemoryLane(@Auth() auth: AuthDto, @Query() dto: MemoryLaneDto): Promise<MemoryLaneResponseDto[]> {
-    return this.service.getMemoryLane(auth, dto);
-  }
-
   @Get('random')
   @Authenticated()
   @EndpointLifecycle({ deprecatedAt: 'v1.116.0' })

server/src/controllers/file-report.controller.ts

@@ -1,29 +0,0 @@
-import { Body, Controller, Get, Post } from '@nestjs/common';
-import { ApiTags } from '@nestjs/swagger';
-import { FileChecksumDto, FileChecksumResponseDto, FileReportDto, FileReportFixDto } from 'src/dtos/audit.dto';
-import { Authenticated } from 'src/middleware/auth.guard';
-import { AuditService } from 'src/services/audit.service';
-
-@ApiTags('File Reports')
-@Controller('reports')
-export class ReportController {
-  constructor(private service: AuditService) {}
-
-  @Get()
-  @Authenticated({ admin: true })
-  getAuditFiles(): Promise<FileReportDto> {
-    return this.service.getFileReport();
-  }
-
-  @Post('checksum')
-  @Authenticated({ admin: true })
-  getFileChecksums(@Body() dto: FileChecksumDto): Promise<FileChecksumResponseDto[]> {
-    return this.service.getChecksums(dto);
-  }
-
-  @Post('fix')
-  @Authenticated({ admin: true })
-  fixAuditFiles(@Body() dto: FileReportFixDto): Promise<void> {
-    return this.service.fixItems(dto.items);
-  }
-}

server/src/controllers/index.ts

@@ -8,7 +8,6 @@ import { AuthController } from 'src/controllers/auth.controller';
 import { DownloadController } from 'src/controllers/download.controller';
 import { DuplicateController } from 'src/controllers/duplicate.controller';
 import { FaceController } from 'src/controllers/face.controller';
-import { ReportController } from 'src/controllers/file-report.controller';
 import { JobController } from 'src/controllers/job.controller';
 import { LibraryController } from 'src/controllers/library.controller';
 import { MapController } from 'src/controllers/map.controller';
@@ -53,7 +52,6 @@ export const controllers = [
   OAuthController,
   PartnerController,
   PersonController,
-  ReportController,
   SearchController,
   ServerController,
   SessionController,

server/src/controllers/search.controller.ts

@@ -46,7 +46,7 @@ export class SearchController {
   @Get('explore')
   @Authenticated()
   getExploreData(@Auth() auth: AuthDto): Promise<SearchExploreResponseDto[]> {
-    return this.service.getExploreData(auth) as Promise<SearchExploreResponseDto[]>;
+    return this.service.getExploreData(auth);
   }
 
   @Get('person')

server/src/dtos/asset-response.dto.ts

@@ -191,10 +191,3 @@ export function mapAsset(entity: MapAsset, options: AssetMapOptions = {}): Asset
     resized: true,
   };
 }
-
-export class MemoryLaneResponseDto {
-  @ApiProperty({ type: 'integer' })
-  yearsAgo!: number;
-
-  assets!: AssetResponseDto[];
-}

server/src/dtos/audit.dto.ts

@@ -1,73 +0,0 @@
-import { ApiProperty } from '@nestjs/swagger';
-import { Type } from 'class-transformer';
-import { IsArray, IsEnum, IsString, IsUUID, ValidateNested } from 'class-validator';
-import { AssetPathType, EntityType, PathType, PersonPathType, UserPathType } from 'src/enum';
-import { Optional, ValidateDate, ValidateUUID } from 'src/validation';
-
-const PathEnum = Object.values({ ...AssetPathType, ...PersonPathType, ...UserPathType });
-
-export class AuditDeletesDto {
-  @ValidateDate()
-  after!: Date;
-
-  @ApiProperty({ enum: EntityType, enumName: 'EntityType' })
-  @IsEnum(EntityType)
-  entityType!: EntityType;
-
-  @Optional()
-  @IsUUID('4')
-  @ApiProperty({ format: 'uuid' })
-  userId?: string;
-}
-
-export enum PathEntityType {
-  ASSET = 'asset',
-  PERSON = 'person',
-  USER = 'user',
-}
-
-export class AuditDeletesResponseDto {
-  needsFullSync!: boolean;
-  ids!: string[];
-}
-
-export class FileReportDto {
-  orphans!: FileReportItemDto[];
-  extras!: string[];
-}
-
-export class FileChecksumDto {
-  @IsString({ each: true })
-  filenames!: string[];
-}
-
-export class FileChecksumResponseDto {
-  filename!: string;
-  checksum!: string;
-}
-
-export class FileReportFixDto {
-  @IsArray()
-  @ValidateNested({ each: true })
-  @Type(() => FileReportItemDto)
-  items!: FileReportItemDto[];
-}
-
-// used both as request and response dto
-export class FileReportItemDto {
-  @ValidateUUID()
-  entityId!: string;
-
-  @ApiProperty({ enumName: 'PathEntityType', enum: PathEntityType })
-  @IsEnum(PathEntityType)
-  entityType!: PathEntityType;
-
-  @ApiProperty({ enumName: 'PathType', enum: PathEnum })
-  @IsEnum(PathEnum)
-  pathType!: PathType;
-
-  @IsString()
-  pathValue!: string;
-
-  checksum?: string;
-}

server/src/dtos/system-config.dto.ts

@@ -25,6 +25,7 @@ import {
   Colorspace,
   ImageFormat,
   LogLevel,
+  OAuthTokenEndpointAuthMethod,
   QueueName,
   ToneMapping,
   TranscodeHWAccel,
@@ -33,7 +34,7 @@ import {
   VideoContainer,
 } from 'src/enum';
 import { ConcurrentQueueName } from 'src/types';
-import { IsCronExpression, ValidateBoolean } from 'src/validation';
+import { IsCronExpression, Optional, ValidateBoolean } from 'src/validation';
 
 const isLibraryScanEnabled = (config: SystemConfigLibraryScanDto) => config.enabled;
 const isOAuthEnabled = (config: SystemConfigOAuthDto) => config.enabled;
@@ -344,10 +345,19 @@ class SystemConfigOAuthDto {
   clientId!: string;
 
   @ValidateIf(isOAuthEnabled)
-  @IsNotEmpty()
   @IsString()
   clientSecret!: string;
 
+  @IsEnum(OAuthTokenEndpointAuthMethod)
+  @ApiProperty({ enum: OAuthTokenEndpointAuthMethod, enumName: 'OAuthTokenEndpointAuthMethod' })
+  tokenEndpointAuthMethod!: OAuthTokenEndpointAuthMethod;
+
+  @IsInt()
+  @IsPositive()
+  @Optional()
+  @ApiProperty({ type: 'integer' })
+  timeout!: number;
+
   @IsNumber()
   @Min(0)
   defaultStorageQuota!: number;

server/src/enum.ts

@@ -605,3 +605,8 @@ export enum NotificationType {
   SystemMessage = 'SystemMessage',
   Custom = 'Custom',
 }
+
+export enum OAuthTokenEndpointAuthMethod {
+  CLIENT_SECRET_POST = 'client_secret_post',
+  CLIENT_SECRET_BASIC = 'client_secret_basic',
+}

server/src/migrations/1700713994428-AddCLIPEmbeddingIndex.ts

@@ -1,5 +1,5 @@
-import { DatabaseExtension } from 'src/enum';
 import { ConfigRepository } from 'src/repositories/config.repository';
+import { vectorIndexQuery } from 'src/utils/database';
 import { MigrationInterface, QueryRunner } from 'typeorm';
 
 const vectorExtension = new ConfigRepository().getEnv().database.vectorExtension;
@@ -8,15 +8,9 @@ export class AddCLIPEmbeddingIndex1700713994428 implements MigrationInterface {
   name = 'AddCLIPEmbeddingIndex1700713994428';
 
   public async up(queryRunner: QueryRunner): Promise<void> {
-    if (vectorExtension === DatabaseExtension.VECTORS) {
-      await queryRunner.query(`SET vectors.pgvector_compatibility=on`);
-    }
     await queryRunner.query(`SET search_path TO "$user", public, vectors`);
 
-    await queryRunner.query(`
-      CREATE INDEX IF NOT EXISTS clip_index ON smart_search
-      USING hnsw (embedding vector_cosine_ops)
-      WITH (ef_construction = 300, m = 16)`);
+    await queryRunner.query(vectorIndexQuery({ vectorExtension, table: 'smart_search', indexName: 'clip_index' }));
   }
 
   public async down(queryRunner: QueryRunner): Promise<void> {

server/src/migrations/1700714033632-AddFaceEmbeddingIndex.ts

@@ -1,5 +1,5 @@
-import { DatabaseExtension } from 'src/enum';
 import { ConfigRepository } from 'src/repositories/config.repository';
+import { vectorIndexQuery } from 'src/utils/database';
 import { MigrationInterface, QueryRunner } from 'typeorm';
 
 const vectorExtension = new ConfigRepository().getEnv().database.vectorExtension;
@@ -8,15 +8,9 @@ export class AddFaceEmbeddingIndex1700714033632 implements MigrationInterface {
   name = 'AddFaceEmbeddingIndex1700714033632';
 
   public async up(queryRunner: QueryRunner): Promise<void> {
-    if (vectorExtension === DatabaseExtension.VECTORS) {
-      await queryRunner.query(`SET vectors.pgvector_compatibility=on`);
-    }
     await queryRunner.query(`SET search_path TO "$user", public, vectors`);
 
-    await queryRunner.query(`
-      CREATE INDEX IF NOT EXISTS face_index ON asset_faces
-      USING hnsw (embedding vector_cosine_ops)
-      WITH (ef_construction = 300, m = 16)`);
+    await queryRunner.query(vectorIndexQuery({ vectorExtension, table: 'asset_faces', indexName: 'face_index' }));
   }
 
   public async down(queryRunner: QueryRunner): Promise<void> {

server/src/migrations/1718486162779-AddFaceSearchRelation.ts

@@ -1,5 +1,6 @@
 import { DatabaseExtension } from 'src/enum';
 import { ConfigRepository } from 'src/repositories/config.repository';
+import { vectorIndexQuery } from 'src/utils/database';
 import { MigrationInterface, QueryRunner } from 'typeorm';
 
 const vectorExtension = new ConfigRepository().getEnv().database.vectorExtension;
@@ -8,7 +9,6 @@ export class AddFaceSearchRelation1718486162779 implements MigrationInterface {
   public async up(queryRunner: QueryRunner): Promise<void> {
     if (vectorExtension === DatabaseExtension.VECTORS) {
       await queryRunner.query(`SET search_path TO "$user", public, vectors`);
-      await queryRunner.query(`SET vectors.pgvector_compatibility=on`);
     }
 
     const hasEmbeddings = async (tableName: string): Promise<boolean> => {
@@ -47,21 +47,14 @@ export class AddFaceSearchRelation1718486162779 implements MigrationInterface {
     await queryRunner.query(`ALTER TABLE face_search ALTER COLUMN embedding SET DATA TYPE real[]`);
     await queryRunner.query(`ALTER TABLE face_search ALTER COLUMN embedding SET DATA TYPE vector(512)`);
 
-    await queryRunner.query(`
-      CREATE INDEX IF NOT EXISTS clip_index ON smart_search
-      USING hnsw (embedding vector_cosine_ops)
-      WITH (ef_construction = 300, m = 16)`);
+    await queryRunner.query(vectorIndexQuery({ vectorExtension, table: 'smart_search', indexName: 'clip_index' }));
 
-    await queryRunner.query(`
-      CREATE INDEX face_index ON face_search
-      USING hnsw (embedding vector_cosine_ops)
-      WITH (ef_construction = 300, m = 16)`);
+    await queryRunner.query(vectorIndexQuery({ vectorExtension, table: 'face_search', indexName: 'face_index' }));
   }
 
   public async down(queryRunner: QueryRunner): Promise<void> {
     if (vectorExtension === DatabaseExtension.VECTORS) {
       await queryRunner.query(`SET search_path TO "$user", public, vectors`);
-      await queryRunner.query(`SET vectors.pgvector_compatibility=on`);
     }
 
     await queryRunner.query(`ALTER TABLE asset_faces ADD COLUMN "embedding" vector(512)`);
@@ -74,9 +67,6 @@ export class AddFaceSearchRelation1718486162779 implements MigrationInterface {
       WHERE id = fs."faceId"`);
     await queryRunner.query(`DROP TABLE face_search`);
 
-    await queryRunner.query(`
-      CREATE INDEX face_index ON asset_faces
-      USING hnsw (embedding vector_cosine_ops)
-      WITH (ef_construction = 300, m = 16)`);
+    await queryRunner.query(vectorIndexQuery({ vectorExtension, table: 'asset_faces', indexName: 'face_index' }));
   }
 }

server/src/queries/asset.job.repository.sql

@@ -194,6 +194,43 @@ where
   "asset_files"."assetId" = $1
   and "asset_files"."type" = $2
 
+-- AssetJobRepository.streamForSearchDuplicates
+select
+  "assets"."id"
+from
+  "assets"
+  inner join "asset_job_status" as "job_status" on "assetId" = "assets"."id"
+where
+  "assets"."isVisible" = $1
+  and "assets"."deletedAt" is null
+  and "job_status"."previewAt" is not null
+  and not exists (
+    select
+    from
+      "smart_search"
+    where
+      "assetId" = "assets"."id"
+  )
+  and "job_status"."duplicatesDetectedAt" is null
+
+-- AssetJobRepository.streamForEncodeClip
+select
+  "assets"."id"
+from
+  "assets"
+  inner join "asset_job_status" as "job_status" on "assetId" = "assets"."id"
+where
+  "assets"."isVisible" = $1
+  and "assets"."deletedAt" is null
+  and "job_status"."previewAt" is not null
+  and not exists (
+    select
+    from
+      "smart_search"
+    where
+      "assetId" = "assets"."id"
+  )
+
 -- AssetJobRepository.getForClipEncoding
 select
   "assets"."id",
@@ -432,3 +469,37 @@ from
   "assets"
 where
   "assets"."deletedAt" <= $1
+
+-- AssetJobRepository.streamForSidecar
+select
+  "assets"."id"
+from
+  "assets"
+where
+  (
+    "assets"."sidecarPath" = $1
+    or "assets"."sidecarPath" is null
+  )
+  and "assets"."isVisible" = $2
+
+-- AssetJobRepository.streamForDetectFacesJob
+select
+  "assets"."id"
+from
+  "assets"
+  inner join "asset_job_status" as "job_status" on "assetId" = "assets"."id"
+where
+  "assets"."isVisible" = $1
+  and "assets"."deletedAt" is null
+  and "job_status"."previewAt" is not null
+  and "job_status"."facesRecognizedAt" is null
+order by
+  "assets"."createdAt" desc
+
+-- AssetJobRepository.streamForMigrationJob
+select
+  "id"
+from
+  "assets"
+where
+  "assets"."deletedAt" is null

server/src/queries/asset.repository.sql

@@ -232,25 +232,6 @@ where
 limit
   $3
 
--- AssetRepository.getWithout (sidecar)
-select
-  "assets".*
-from
-  "assets"
-where
-  (
-    "assets"."sidecarPath" = $1
-    or "assets"."sidecarPath" is null
-  )
-  and "assets"."isVisible" = $2
-  and "deletedAt" is null
-order by
-  "createdAt"
-limit
-  $3
-offset
-  $4
-
 -- AssetRepository.getTimeBuckets
 with
   "assets" as (

server/src/repositories/asset-job.repository.ts

@@ -135,6 +135,36 @@ export class AssetJobRepository {
       .execute();
   }
 
+  private assetsWithPreviews() {
+    return this.db
+      .selectFrom('assets')
+      .where('assets.isVisible', '=', true)
+      .where('assets.deletedAt', 'is', null)
+      .innerJoin('asset_job_status as job_status', 'assetId', 'assets.id')
+      .where('job_status.previewAt', 'is not', null);
+  }
+
+  @GenerateSql({ params: [], stream: true })
+  streamForSearchDuplicates(force?: boolean) {
+    return this.assetsWithPreviews()
+      .where((eb) => eb.not((eb) => eb.exists(eb.selectFrom('smart_search').whereRef('assetId', '=', 'assets.id'))))
+      .$if(!force, (qb) => qb.where('job_status.duplicatesDetectedAt', 'is', null))
+      .select(['assets.id'])
+      .stream();
+  }
+
+  @GenerateSql({ params: [], stream: true })
+  streamForEncodeClip(force?: boolean) {
+    return this.assetsWithPreviews()
+      .select(['assets.id'])
+      .$if(!force, (qb) =>
+        qb.where((eb) =>
+          eb.not((eb) => eb.exists(eb.selectFrom('smart_search').whereRef('assetId', '=', 'assets.id'))),
+        ),
+      )
+      .stream();
+  }
+
   @GenerateSql({ params: [DummyValue.UUID] })
   getForClipEncoding(id: string) {
     return this.db
@@ -292,4 +322,30 @@ export class AssetJobRepository {
       .where('assets.deletedAt', '<=', trashedBefore)
       .stream();
   }
+
+  @GenerateSql({ params: [], stream: true })
+  streamForSidecar(force?: boolean) {
+    return this.db
+      .selectFrom('assets')
+      .select(['assets.id'])
+      .$if(!force, (qb) =>
+        qb.where((eb) => eb.or([eb('assets.sidecarPath', '=', ''), eb('assets.sidecarPath', 'is', null)])),
+      )
+      .where('assets.isVisible', '=', true)
+      .stream();
+  }
+
+  @GenerateSql({ params: [], stream: true })
+  streamForDetectFacesJob(force?: boolean) {
+    return this.assetsWithPreviews()
+      .$if(!force, (qb) => qb.where('job_status.facesRecognizedAt', 'is', null))
+      .select(['assets.id'])
+      .orderBy('assets.createdAt', 'desc')
+      .stream();
+  }
+
+  @GenerateSql({ params: [DummyValue.DATE], stream: true })
+  streamForMigrationJob() {
+    return this.db.selectFrom('assets').select(['id']).where('assets.deletedAt', 'is', null).stream();
+  }
 }

server/src/repositories/asset.repository.ts

@@ -7,14 +7,12 @@ import { AssetFiles, AssetJobStatus, Assets, DB, Exif } from 'src/db';
 import { Chunked, ChunkedArray, DummyValue, GenerateSql } from 'src/decorators';
 import { MapAsset } from 'src/dtos/asset-response.dto';
 import { AssetFileType, AssetOrder, AssetStatus, AssetType } from 'src/enum';
-import { AssetSearchOptions, SearchExploreItem, SearchExploreItemSet } from 'src/repositories/search.repository';
 import {
   anyUuid,
   asUuid,
   hasPeople,
   hasPeopleNoJoin,
   removeUndefinedKeys,
-  searchAssetBuilder,
   truncatedDate,
   unnest,
   withExif,
@@ -29,7 +27,7 @@ import {
   withTags,
 } from 'src/utils/database';
 import { globToSqlPattern } from 'src/utils/misc';
-import { PaginationOptions, paginationHelper } from 'src/utils/pagination';
+import { PaginationOptions } from 'src/utils/pagination';
 
 export type AssetStats = Record<AssetType, number>;
 
@@ -47,16 +45,6 @@ export interface LivePhotoSearchOptions {
   type: AssetType;
 }
 
-export enum WithoutProperty {
-  THUMBNAIL = 'thumbnail',
-  ENCODED_VIDEO = 'encoded-video',
-  EXIF = 'exif',
-  SMART_SEARCH = 'smart-search',
-  DUPLICATE = 'duplicate',
-  FACES = 'faces',
-  SIDECAR = 'sidecar',
-}
-
 export enum WithProperty {
   SIDECAR = 'sidecar',
 }
@@ -337,10 +325,6 @@ export class AssetRepository {
     return assets.map((asset) => asset.deviceAssetId);
   }
 
-  getByUserId(pagination: PaginationOptions, userId: string, options: Omit<AssetSearchOptions, 'userIds'> = {}) {
-    return this.getAll(pagination, { ...options, userIds: [userId] });
-  }
-
   @GenerateSql({ params: [DummyValue.UUID, DummyValue.STRING] })
   getByLibraryIdAndOriginalPath(libraryId: string, originalPath: string) {
     return this.db
@@ -352,16 +336,6 @@ export class AssetRepository {
       .executeTakeFirst();
   }
 
-  async getAll(pagination: PaginationOptions, { orderDirection, ...options }: AssetSearchOptions = {}) {
-    const builder = searchAssetBuilder(this.db, options)
-      .select(withFiles)
-      .orderBy('assets.createdAt', orderDirection ?? 'asc')
-      .limit(pagination.take + 1)
-      .offset(pagination.skip ?? 0);
-    const items = await builder.execute();
-    return paginationHelper(items, pagination.take);
-  }
-
   /**
    * Get assets by device's Id on the database
    * @param ownerId
@@ -531,77 +505,6 @@ export class AssetRepository {
       .executeTakeFirst();
   }
 
-  @GenerateSql(
-    ...Object.values(WithProperty).map((property) => ({
-      name: property,
-      params: [DummyValue.PAGINATION, property],
-    })),
-  )
-  async getWithout(pagination: PaginationOptions, property: WithoutProperty) {
-    const items = await this.db
-      .selectFrom('assets')
-      .selectAll('assets')
-      .$if(property === WithoutProperty.DUPLICATE, (qb) =>
-        qb
-          .innerJoin('asset_job_status as job_status', 'assets.id', 'job_status.assetId')
-          .where('job_status.duplicatesDetectedAt', 'is', null)
-          .where('job_status.previewAt', 'is not', null)
-          .where((eb) => eb.exists(eb.selectFrom('smart_search').where('assetId', '=', eb.ref('assets.id'))))
-          .where('assets.isVisible', '=', true),
-      )
-      .$if(property === WithoutProperty.ENCODED_VIDEO, (qb) =>
-        qb
-          .where('assets.type', '=', AssetType.VIDEO)
-          .where((eb) => eb.or([eb('assets.encodedVideoPath', 'is', null), eb('assets.encodedVideoPath', '=', '')])),
-      )
-      .$if(property === WithoutProperty.EXIF, (qb) =>
-        qb
-          .leftJoin('asset_job_status as job_status', 'assets.id', 'job_status.assetId')
-          .where((eb) => eb.or([eb('job_status.metadataExtractedAt', 'is', null), eb('assetId', 'is', null)]))
-          .where('assets.isVisible', '=', true),
-      )
-      .$if(property === WithoutProperty.FACES, (qb) =>
-        qb
-          .innerJoin('asset_job_status as job_status', 'assetId', 'assets.id')
-          .where('job_status.previewAt', 'is not', null)
-          .where('job_status.facesRecognizedAt', 'is', null)
-          .where('assets.isVisible', '=', true),
-      )
-      .$if(property === WithoutProperty.SIDECAR, (qb) =>
-        qb
-          .where((eb) => eb.or([eb('assets.sidecarPath', '=', ''), eb('assets.sidecarPath', 'is', null)]))
-          .where('assets.isVisible', '=', true),
-      )
-      .$if(property === WithoutProperty.SMART_SEARCH, (qb) =>
-        qb
-          .innerJoin('asset_job_status as job_status', 'assetId', 'assets.id')
-          .where('job_status.previewAt', 'is not', null)
-          .where('assets.isVisible', '=', true)
-          .where((eb) =>
-            eb.not((eb) => eb.exists(eb.selectFrom('smart_search').whereRef('assetId', '=', 'assets.id'))),
-          ),
-      )
-      .$if(property === WithoutProperty.THUMBNAIL, (qb) =>
-        qb
-          .innerJoin('asset_job_status as job_status', 'assetId', 'assets.id')
-          .where('assets.isVisible', '=', true)
-          .where((eb) =>
-            eb.or([
-              eb('job_status.previewAt', 'is', null),
-              eb('job_status.thumbnailAt', 'is', null),
-              eb('assets.thumbhash', 'is', null),
-            ]),
-          ),
-      )
-      .where('deletedAt', 'is', null)
-      .limit(pagination.take + 1)
-      .offset(pagination.skip ?? 0)
-      .orderBy('createdAt')
-      .execute();
-
-    return paginationHelper(items, pagination.take);
-  }
-
   getStatistics(ownerId: string, { isArchived, isFavorite, isTrashed }: AssetStatsOptions): Promise<AssetStats> {
     return this.db
       .selectFrom('assets')
@@ -816,10 +719,7 @@ export class AssetRepository {
   }
 
   @GenerateSql({ params: [DummyValue.UUID, { minAssetsPerField: 5, maxFields: 12 }] })
-  async getAssetIdByCity(
-    ownerId: string,
-    { minAssetsPerField, maxFields }: AssetExploreFieldOptions,
-  ): Promise<SearchExploreItem<string>> {
+  async getAssetIdByCity(ownerId: string, { minAssetsPerField, maxFields }: AssetExploreFieldOptions) {
     const items = await this.db
       .with('cities', (qb) =>
         qb
@@ -834,6 +734,7 @@ export class AssetRepository {
       .innerJoin('cities', 'exif.city', 'cities.city')
       .distinctOn('exif.city')
       .select(['assetId as data', 'exif.city as value'])
+      .$narrowType<{ value: NotNull }>()
       .where('ownerId', '=', asUuid(ownerId))
       .where('isVisible', '=', true)
       .where('isArchived', '=', false)
@@ -842,7 +743,7 @@ export class AssetRepository {
       .limit(maxFields)
       .execute();
 
-    return { fieldName: 'exifInfo.city', items: items as SearchExploreItemSet<string> };
+    return { fieldName: 'exifInfo.city', items };
   }
 
   @GenerateSql({

server/src/repositories/database.repository.ts

@@ -12,6 +12,7 @@ import { DatabaseExtension, DatabaseLock, VectorIndex } from 'src/enum';
 import { ConfigRepository } from 'src/repositories/config.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { ExtensionVersion, VectorExtension, VectorUpdateResult } from 'src/types';
+import { vectorIndexQuery } from 'src/utils/database';
 import { isValidInteger } from 'src/validation';
 import { DataSource } from 'typeorm';
 
@@ -119,12 +120,7 @@ export class DatabaseRepository {
         await sql`ALTER TABLE ${sql.raw(table)} ALTER COLUMN embedding SET DATA TYPE vector(${sql.raw(String(dimSize))})`.execute(
           tx,
         );
-        await sql`SET vectors.pgvector_compatibility=on`.execute(tx);
-        await sql`
-          CREATE INDEX IF NOT EXISTS ${sql.raw(index)} ON ${sql.raw(table)}
-          USING hnsw (embedding vector_cosine_ops)
-          WITH (ef_construction = 300, m = 16)
-        `.execute(tx);
+        await sql.raw(vectorIndexQuery({ vectorExtension: this.vectorExtension, table, indexName: index })).execute(tx);
       });
     }
   }

server/src/repositories/event.repository.ts

@@ -19,7 +19,7 @@ import { ReleaseNotification, ServerVersionResponseDto } from 'src/dtos/server.d
 import { ImmichWorker, MetadataKey, QueueName } from 'src/enum';
 import { ConfigRepository } from 'src/repositories/config.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
-import { JobItem } from 'src/types';
+import { JobItem, JobSource } from 'src/types';
 import { handlePromiseError } from 'src/utils/misc';
 
 type EmitHandlers = Partial<{ [T in EmitEvent]: Array<EventItem<T>> }>;
@@ -48,7 +48,7 @@ type EventMap = {
   'config.validate': [{ newConfig: SystemConfig; oldConfig: SystemConfig }];
 
   // album events
-  'album.update': [{ id: string; recipientIds: string[] }];
+  'album.update': [{ id: string; recipientId: string }];
   'album.invite': [{ id: string; userId: string }];
 
   // asset events
@@ -58,6 +58,7 @@ type EventMap = {
   'asset.show': [{ assetId: string; userId: string }];
   'asset.trash': [{ assetId: string; userId: string }];
   'asset.delete': [{ assetId: string; userId: string }];
+  'asset.metadataExtracted': [{ assetId: string; userId: string; source?: JobSource }];
 
   // asset bulk events
   'assets.trash': [{ assetIds: string[]; userId: string }];

server/src/repositories/job.repository.ts

@@ -9,7 +9,7 @@ import { JobName, JobStatus, MetadataKey, QueueCleanType, QueueName } from 'src/
 import { ConfigRepository } from 'src/repositories/config.repository';
 import { EventRepository } from 'src/repositories/event.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
-import { IEntityJob, JobCounts, JobItem, JobOf, QueueStatus } from 'src/types';
+import { JobCounts, JobItem, JobOf, QueueStatus } from 'src/types';
 import { getKeyByValue, getMethodNames, ImmichStartupError } from 'src/utils/misc';
 
 type JobMapItem = {
@@ -206,7 +206,10 @@ export class JobRepository {
   private getJobOptions(item: JobItem): JobsOptions | null {
     switch (item.name) {
       case JobName.NOTIFY_ALBUM_UPDATE: {
-        return { jobId: item.data.id, delay: item.data?.delay };
+        return {
+          jobId: `${item.data.id}/${item.data.recipientId}`,
+          delay: item.data?.delay,
+        };
       }
       case JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE: {
         return { jobId: item.data.id };
@@ -227,19 +230,12 @@ export class JobRepository {
     return this.moduleRef.get<Queue>(getQueueToken(queue), { strict: false });
   }
 
-  public async removeJob(jobId: string, name: JobName): Promise<IEntityJob | undefined> {
-    const existingJob = await this.getQueue(this.getQueueName(name)).getJob(jobId);
-    if (!existingJob) {
-      return;
-    }
-    try {
+  /** @deprecated */
+  // todo: remove this when asset notifications no longer need it.
+  public async removeJob(name: JobName, jobID: string): Promise<void> {
+    const existingJob = await this.getQueue(this.getQueueName(name)).getJob(jobID);
+    if (existingJob) {
       await existingJob.remove();
-    } catch (error: any) {
-      if (error.message?.includes('Missing key for job')) {
-        return;
-      }
-      throw error;
     }
-    return existingJob.data;
   }
 }

server/src/repositories/logging.repository.ts

@@ -5,7 +5,7 @@ import { Telemetry } from 'src/decorators';
 import { LogLevel } from 'src/enum';
 import { ConfigRepository } from 'src/repositories/config.repository';
 
-type LogDetails = any[];
+type LogDetails = any;
 type LogFunction = () => string;
 
 const LOG_LEVELS = [LogLevel.VERBOSE, LogLevel.DEBUG, LogLevel.LOG, LogLevel.WARN, LogLevel.ERROR, LogLevel.FATAL];

server/src/repositories/oauth.repository.ts

@@ -1,16 +1,19 @@
 import { Injectable, InternalServerErrorException } from '@nestjs/common';
 import type { UserInfoResponse } from 'openid-client' with { 'resolution-mode': 'import' };
+import { OAuthTokenEndpointAuthMethod } from 'src/enum';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 
 export type OAuthConfig = {
   clientId: string;
-  clientSecret: string;
+  clientSecret?: string;
   issuerUrl: string;
   mobileOverrideEnabled: boolean;
   mobileRedirectUri: string;
   profileSigningAlgorithm: string;
   scope: string;
   signingAlgorithm: string;
+  tokenEndpointAuthMethod: OAuthTokenEndpointAuthMethod;
+  timeout: number;
 };
 export type OAuthProfile = UserInfoResponse;
 
@@ -76,12 +79,10 @@ export class OAuthRepository {
         );
       }
 
-      if (error.code === 'OAUTH_INVALID_RESPONSE') {
-        this.logger.warn(`Invalid response from authorization server. Cause: ${error.cause?.message}`);
-        throw error.cause;
-      }
+      this.logger.error(`OAuth login failed: ${error.message}`);
+      this.logger.error(error);
 
-      throw error;
+      throw new Error('OAuth login failed', { cause: error });
     }
   }
 
@@ -103,6 +104,8 @@ export class OAuthRepository {
     clientSecret,
     profileSigningAlgorithm,
     signingAlgorithm,
+    tokenEndpointAuthMethod,
+    timeout,
   }: OAuthConfig) {
     try {
       const { allowInsecureRequests, discovery } = await import('openid-client');
@@ -114,14 +117,38 @@ export class OAuthRepository {
           response_types: ['code'],
           userinfo_signed_response_alg: profileSigningAlgorithm === 'none' ? undefined : profileSigningAlgorithm,
           id_token_signed_response_alg: signingAlgorithm,
-          timeout: 30_000,
         },
-        undefined,
-        { execute: [allowInsecureRequests] },
+        await this.getTokenAuthMethod(tokenEndpointAuthMethod, clientSecret),
+        {
+          execute: [allowInsecureRequests],
+          timeout,
+        },
       );
     } catch (error: any | AggregateError) {
       this.logger.error(`Error in OAuth discovery: ${error}`, error?.stack, error?.errors);
       throw new InternalServerErrorException(`Error in OAuth discovery: ${error}`, { cause: error });
     }
   }
+
+  private async getTokenAuthMethod(tokenEndpointAuthMethod: OAuthTokenEndpointAuthMethod, clientSecret?: string) {
+    const { None, ClientSecretPost, ClientSecretBasic } = await import('openid-client');
+
+    if (!clientSecret) {
+      return None();
+    }
+
+    switch (tokenEndpointAuthMethod) {
+      case OAuthTokenEndpointAuthMethod.CLIENT_SECRET_POST: {
+        return ClientSecretPost(clientSecret);
+      }
+
+      case OAuthTokenEndpointAuthMethod.CLIENT_SECRET_BASIC: {
+        return ClientSecretBasic(clientSecret);
+      }
+
+      default: {
+        return None();
+      }
+    }
+  }
 }

server/src/repositories/person.repository.ts

@@ -6,7 +6,7 @@ import { AssetFaces, DB, FaceSearch, Person } from 'src/db';
 import { ChunkedArray, DummyValue, GenerateSql } from 'src/decorators';
 import { AssetFileType, SourceType } from 'src/enum';
 import { removeUndefinedKeys } from 'src/utils/database';
-import { PaginationOptions } from 'src/utils/pagination';
+import { paginationHelper, PaginationOptions } from 'src/utils/pagination';
 
 export interface PersonSearchOptions {
   minimumFaceCount: number;
@@ -200,11 +200,7 @@ export class PersonRepository {
       .limit(pagination.take + 1)
       .execute();
 
-    if (items.length > pagination.take) {
-      return { items: items.slice(0, -1), hasNextPage: true };
-    }
-
-    return { items, hasNextPage: false };
+    return paginationHelper(items, pagination.take);
   }
 
   @GenerateSql()

server/src/repositories/search.repository.ts

@@ -6,42 +6,12 @@ import { DB, Exif } from 'src/db';
 import { DummyValue, GenerateSql } from 'src/decorators';
 import { MapAsset } from 'src/dtos/asset-response.dto';
 import { AssetStatus, AssetType } from 'src/enum';
-import { anyUuid, asUuid, searchAssetBuilder } from 'src/utils/database';
+import { ConfigRepository } from 'src/repositories/config.repository';
+import { anyUuid, asUuid, searchAssetBuilder, vectorIndexQuery } from 'src/utils/database';
+import { paginationHelper } from 'src/utils/pagination';
 import { isValidInteger } from 'src/validation';
 
-export interface SearchResult<T> {
-  /** total matches */
-  total: number;
-  /** collection size */
-  count: number;
-  /** current page */
-  page: number;
-  /** items for page */
-  items: T[];
-  /** score */
-  distances: number[];
-  facets: SearchFacet[];
-}
-
-export interface SearchFacet {
-  fieldName: string;
-  counts: Array<{
-    count: number;
-    value: string;
-  }>;
-}
-
-export type SearchExploreItemSet<T> = Array<{
-  value: string;
-  data: T;
-}>;
-
-export interface SearchExploreItem<T> {
-  fieldName: string;
-  items: SearchExploreItemSet<T>;
-}
-
-export interface SearchAssetIDOptions {
+export interface SearchAssetIdOptions {
   checksum?: Buffer;
   deviceAssetId?: string;
   id?: string;
@@ -53,7 +23,7 @@ export interface SearchUserIdOptions {
   userIds?: string[];
 }
 
-export type SearchIdOptions = SearchAssetIDOptions & SearchUserIdOptions;
+export type SearchIdOptions = SearchAssetIdOptions & SearchUserIdOptions;
 
 export interface SearchStatusOptions {
   isArchived?: boolean;
@@ -143,8 +113,6 @@ type BaseAssetSearchOptions = SearchDateOptions &
 
 export type AssetSearchOptions = BaseAssetSearchOptions & SearchRelationOptions;
 
-export type AssetSearchOneToOneRelationOptions = BaseAssetSearchOptions & SearchOneToOneRelationOptions;
-
 export type AssetSearchBuilderOptions = Omit<AssetSearchOptions, 'orderDirection'>;
 
 export type SmartSearchOptions = SearchDateOptions &
@@ -201,7 +169,10 @@ export interface GetCameraMakesOptions {
 
 @Injectable()
 export class SearchRepository {
-  constructor(@InjectKysely() private db: Kysely<DB>) {}
+  constructor(
+    @InjectKysely() private db: Kysely<DB>,
+    private configRepository: ConfigRepository,
+  ) {}
 
   @GenerateSql({
     params: [
@@ -222,9 +193,8 @@ export class SearchRepository {
       .limit(pagination.size + 1)
       .offset((pagination.page - 1) * pagination.size)
       .execute();
-    const hasNextPage = items.length > pagination.size;
-    items.splice(pagination.size);
-    return { items, hasNextPage };
+
+    return paginationHelper(items, pagination.size);
   }
 
   @GenerateSql({
@@ -279,9 +249,7 @@ export class SearchRepository {
       .offset((pagination.page - 1) * pagination.size)
       .execute();
 
-    const hasNextPage = items.length > pagination.size;
-    items.splice(pagination.size);
-    return { items, hasNextPage };
+    return paginationHelper(items, pagination.size);
   }
 
   @GenerateSql({
@@ -446,8 +414,8 @@ export class SearchRepository {
   async upsert(assetId: string, embedding: string): Promise<void> {
     await this.db
       .insertInto('smart_search')
-      .values({ assetId: asUuid(assetId), embedding } as any)
-      .onConflict((oc) => oc.column('assetId').doUpdateSet({ embedding } as any))
+      .values({ assetId, embedding })
+      .onConflict((oc) => oc.column('assetId').doUpdateSet((eb) => ({ embedding: eb.ref('excluded.embedding') })))
       .execute();
   }
 
@@ -469,19 +437,32 @@ export class SearchRepository {
     return dimSize;
   }
 
-  setDimensionSize(dimSize: number): Promise<void> {
+  async setDimensionSize(dimSize: number): Promise<void> {
     if (!isValidInteger(dimSize, { min: 1, max: 2 ** 16 })) {
       throw new Error(`Invalid CLIP dimension size: ${dimSize}`);
     }
 
-    return this.db.transaction().execute(async (trx) => {
-      await sql`truncate ${sql.table('smart_search')}`.execute(trx);
+    // this is done in two transactions to handle concurrent writes
+    await this.db.transaction().execute(async (trx) => {
+      await sql`delete from ${sql.table('smart_search')}`.execute(trx);
+      await trx.schema.alterTable('smart_search').dropConstraint('dim_size_constraint').ifExists().execute();
+      await sql`alter table ${sql.table('smart_search')} add constraint dim_size_constraint check (array_length(embedding::real[], 1) = ${sql.lit(dimSize)})`.execute(
+        trx,
+      );
+    });
+
+    const vectorExtension = this.configRepository.getEnv().database.vectorExtension;
+    await this.db.transaction().execute(async (trx) => {
+      await sql`drop index if exists clip_index`.execute(trx);
       await trx.schema
         .alterTable('smart_search')
         .alterColumn('embedding', (col) => col.setDataType(sql.raw(`vector(${dimSize})`)))
         .execute();
-      await sql`reindex index clip_index`.execute(trx);
+      await sql.raw(vectorIndexQuery({ vectorExtension, table: 'smart_search', indexName: 'clip_index' })).execute(trx);
+      await trx.schema.alterTable('smart_search').dropConstraint('dim_size_constraint').ifExists().execute();
     });
+
+    await sql`vacuum analyze ${sql.table('smart_search')}`.execute(this.db);
   }
 
   async deleteAllSearchEmbeddings(): Promise<void> {

server/src/schema/index.ts

@@ -47,14 +47,8 @@ import { UserTable } from 'src/schema/tables/user.table';
 import { VersionHistoryTable } from 'src/schema/tables/version-history.table';
 import { ConfigurationParameter, Database, Extensions } from 'src/sql-tools';
 
-@Extensions(['uuid-ossp', 'unaccent', 'cube', 'earthdistance', 'pg_trgm', 'vectors', 'plpgsql'])
+@Extensions(['uuid-ossp', 'unaccent', 'cube', 'earthdistance', 'pg_trgm', 'plpgsql'])
 @ConfigurationParameter({ name: 'search_path', value: () => '"$user", public, vectors', scope: 'database' })
-@ConfigurationParameter({
-  name: 'vectors.pgvector_compatibility',
-  value: () => 'on',
-  scope: 'user',
-  synchronize: false,
-})
 @Database({ name: 'immich' })
 export class ImmichDatabase {
   tables = [

server/src/schema/migrations/1744910873969-InitialMigration.ts

@@ -2,6 +2,7 @@ import { Kysely, sql } from 'kysely';
 import { DatabaseExtension } from 'src/enum';
 import { ConfigRepository } from 'src/repositories/config.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
+import { vectorIndexQuery } from 'src/utils/database';
 
 const vectorExtension = new ConfigRepository().getEnv().database.vectorExtension;
 const lastMigrationSql = sql<{ name: string }>`SELECT "name" FROM "migrations" ORDER BY "timestamp" DESC LIMIT 1;`;
@@ -29,7 +30,7 @@ export async function up(db: Kysely<any>): Promise<void> {
   await sql`CREATE EXTENSION IF NOT EXISTS "cube";`.execute(db);
   await sql`CREATE EXTENSION IF NOT EXISTS "earthdistance";`.execute(db);
   await sql`CREATE EXTENSION IF NOT EXISTS "pg_trgm";`.execute(db);
-  await sql`CREATE EXTENSION IF NOT EXISTS "vectors";`.execute(db);
+  await sql`CREATE EXTENSION IF NOT EXISTS ${sql.raw(vectorExtension)}`.execute(db);
   await sql`CREATE OR REPLACE FUNCTION immich_uuid_v7(p_timestamp timestamp with time zone default clock_timestamp())
   RETURNS uuid
   VOLATILE LANGUAGE SQL
@@ -108,7 +109,6 @@ export async function up(db: Kysely<any>): Promise<void> {
   $$;`.execute(db);
   if (vectorExtension === DatabaseExtension.VECTORS) {
     await sql`SET search_path TO "$user", public, vectors`.execute(db);
-    await sql`SET vectors.pgvector_compatibility=on`.execute(db);
   }
   await sql`CREATE TYPE "assets_status_enum" AS ENUM ('active','trashed','deleted');`.execute(db);
   await sql`CREATE TYPE "sourcetype" AS ENUM ('machine-learning','exif','manual');`.execute(db);
@@ -293,7 +293,7 @@ export async function up(db: Kysely<any>): Promise<void> {
   await sql`CREATE INDEX "IDX_live_photo_cid" ON "exif" ("livePhotoCID")`.execute(db);
   await sql`CREATE INDEX "IDX_auto_stack_id" ON "exif" ("autoStackId")`.execute(db);
   await sql`CREATE INDEX "IDX_asset_exif_update_id" ON "exif" ("updateId")`.execute(db);
-  await sql`CREATE INDEX "face_index" ON "face_search" USING hnsw (embedding vector_cosine_ops) WITH (ef_construction = 300, m = 16)`.execute(db);
+  await sql.raw(vectorIndexQuery({ vectorExtension, table: 'face_search', indexName: 'face_index' })).execute(db);
   await sql`CREATE INDEX "IDX_geodata_gist_earthcoord" ON "geodata_places" (ll_to_earth_public(latitude, longitude))`.execute(db);
   await sql`CREATE INDEX "idx_geodata_places_name" ON "geodata_places" USING gin (f_unaccent("name") gin_trgm_ops)`.execute(db);
   await sql`CREATE INDEX "idx_geodata_places_admin2_name" ON "geodata_places" USING gin (f_unaccent("admin2Name") gin_trgm_ops)`.execute(db);
@@ -316,7 +316,7 @@ export async function up(db: Kysely<any>): Promise<void> {
   await sql`CREATE INDEX "IDX_sharedlink_albumId" ON "shared_links" ("albumId")`.execute(db);
   await sql`CREATE INDEX "IDX_5b7decce6c8d3db9593d6111a6" ON "shared_link__asset" ("assetsId")`.execute(db);
   await sql`CREATE INDEX "IDX_c9fab4aa97ffd1b034f3d6581a" ON "shared_link__asset" ("sharedLinksId")`.execute(db);
-  await sql`CREATE INDEX "clip_index" ON "smart_search" USING hnsw (embedding vector_cosine_ops) WITH (ef_construction = 300, m = 16)`.execute(db);
+  await sql.raw(vectorIndexQuery({ vectorExtension, table: 'smart_search', indexName: 'clip_index' })).execute(db);
   await sql`CREATE INDEX "IDX_d8ddd9d687816cc490432b3d4b" ON "session_sync_checkpoints" ("sessionId")`.execute(db);
   await sql`CREATE INDEX "IDX_session_sync_checkpoints_update_id" ON "session_sync_checkpoints" ("updateId")`.execute(db);
   await sql`CREATE INDEX "IDX_92e67dc508c705dd66c9461557" ON "tags" ("userId")`.execute(db);

server/src/services/album.service.spec.ts

@@ -606,7 +606,7 @@ describe(AlbumService.name, () => {
       expect(mocks.album.addAssetIds).toHaveBeenCalledWith('album-123', ['asset-1', 'asset-2', 'asset-3']);
       expect(mocks.event.emit).toHaveBeenCalledWith('album.update', {
         id: 'album-123',
-        recipientIds: ['admin_id'],
+        recipientId: 'admin_id',
       });
     });
 

server/src/services/album.service.ts

@@ -170,8 +170,8 @@ export class AlbumService extends BaseService {
         (userId) => userId !== auth.user.id,
       );
 
-      if (allUsersExceptUs.length > 0) {
-        await this.eventRepository.emit('album.update', { id, recipientIds: allUsersExceptUs });
+      for (const recipientId of allUsersExceptUs) {
+        await this.eventRepository.emit('album.update', { id, recipientId });
       }
     }
 

server/src/services/asset.service.spec.ts

@@ -1,6 +1,6 @@
 import { BadRequestException } from '@nestjs/common';
 import { DateTime } from 'luxon';
-import { MapAsset, mapAsset } from 'src/dtos/asset-response.dto';
+import { MapAsset } from 'src/dtos/asset-response.dto';
 import { AssetJobName, AssetStatsResponseDto } from 'src/dtos/asset.dto';
 import { AssetStatus, AssetType, JobName, JobStatus } from 'src/enum';
 import { AssetStats } from 'src/repositories/asset.repository';
@@ -11,7 +11,6 @@ import { faceStub } from 'test/fixtures/face.stub';
 import { userStub } from 'test/fixtures/user.stub';
 import { factory } from 'test/small.factory';
 import { makeStream, newTestService, ServiceMocks } from 'test/utils';
-import { vitest } from 'vitest';
 
 const stats: AssetStats = {
   [AssetType.IMAGE]: 10,
@@ -44,62 +43,6 @@ describe(AssetService.name, () => {
     mockGetById([assetStub.livePhotoStillAsset, assetStub.livePhotoMotionAsset]);
   });
 
-  describe('getMemoryLane', () => {
-    beforeAll(() => {
-      vitest.useFakeTimers();
-      vitest.setSystemTime(new Date('2024-01-15'));
-    });
-
-    afterAll(() => {
-      vitest.useRealTimers();
-    });
-
-    it('should group the assets correctly', async () => {
-      const image1 = { ...assetStub.image, localDateTime: new Date(2023, 1, 15, 0, 0, 0) };
-      const image2 = { ...assetStub.image, localDateTime: new Date(2023, 1, 15, 1, 0, 0) };
-      const image3 = { ...assetStub.image, localDateTime: new Date(2015, 1, 15) };
-      const image4 = { ...assetStub.image, localDateTime: new Date(2009, 1, 15) };
-
-      mocks.partner.getAll.mockResolvedValue([]);
-      mocks.asset.getByDayOfYear.mockResolvedValue([
-        {
-          year: 2023,
-          assets: [image1, image2],
-        },
-        {
-          year: 2015,
-          assets: [image3],
-        },
-        {
-          year: 2009,
-          assets: [image4],
-        },
-      ] as any);
-
-      await expect(sut.getMemoryLane(authStub.admin, { day: 15, month: 1 })).resolves.toEqual([
-        { yearsAgo: 1, title: '1 year ago', assets: [mapAsset(image1), mapAsset(image2)] },
-        { yearsAgo: 9, title: '9 years ago', assets: [mapAsset(image3)] },
-        { yearsAgo: 15, title: '15 years ago', assets: [mapAsset(image4)] },
-      ]);
-
-      expect(mocks.asset.getByDayOfYear.mock.calls).toEqual([[[authStub.admin.user.id], { day: 15, month: 1 }]]);
-    });
-
-    it('should get memories with partners with inTimeline enabled', async () => {
-      const partner = factory.partner();
-      const auth = factory.auth({ user: { id: partner.sharedWithId } });
-
-      mocks.partner.getAll.mockResolvedValue([partner]);
-      mocks.asset.getByDayOfYear.mockResolvedValue([]);
-
-      await sut.getMemoryLane(auth, { day: 15, month: 1 });
-
-      expect(mocks.asset.getByDayOfYear.mock.calls).toEqual([
-        [[auth.user.id, partner.sharedById], { day: 15, month: 1 }],
-      ]);
-    });
-  });
-
   describe('getStatistics', () => {
     it('should get the statistics for a user, excluding archived assets', async () => {
       mocks.asset.getStatistics.mockResolvedValue(stats);

server/src/services/asset.service.ts

@@ -3,13 +3,7 @@ import _ from 'lodash';
 import { DateTime, Duration } from 'luxon';
 import { JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
 import { OnJob } from 'src/decorators';
-import {
-  AssetResponseDto,
-  MapAsset,
-  MemoryLaneResponseDto,
-  SanitizedAssetResponseDto,
-  mapAsset,
-} from 'src/dtos/asset-response.dto';
+import { AssetResponseDto, MapAsset, SanitizedAssetResponseDto, mapAsset } from 'src/dtos/asset-response.dto';
 import {
   AssetBulkDeleteDto,
   AssetBulkUpdateDto,
@@ -20,7 +14,6 @@ import {
   mapStats,
 } from 'src/dtos/asset.dto';
 import { AuthDto } from 'src/dtos/auth.dto';
-import { MemoryLaneDto } from 'src/dtos/search.dto';
 import { AssetStatus, JobName, JobStatus, Permission, QueueName } from 'src/enum';
 import { BaseService } from 'src/services/base.service';
 import { ISidecarWriteJob, JobItem, JobOf } from 'src/types';
@@ -28,26 +21,6 @@ import { getAssetFiles, getMyPartnerIds, onAfterUnlink, onBeforeLink, onBeforeUn
 
 @Injectable()
 export class AssetService extends BaseService {
-  async getMemoryLane(auth: AuthDto, dto: MemoryLaneDto): Promise<MemoryLaneResponseDto[]> {
-    const partnerIds = await getMyPartnerIds({
-      userId: auth.user.id,
-      repository: this.partnerRepository,
-      timelineEnabled: true,
-    });
-    const userIds = [auth.user.id, ...partnerIds];
-
-    const groups = await this.assetRepository.getByDayOfYear(userIds, dto);
-    return groups.map(({ year, assets }) => {
-      const yearsAgo = DateTime.utc().year - year;
-      return {
-        yearsAgo,
-        // TODO move this to clients
-        title: `${yearsAgo} year${yearsAgo > 1 ? 's' : ''} ago`,
-        assets: assets.map((asset) => mapAsset(asset, { auth })),
-      };
-    });
-  }
-
   async getStatistics(auth: AuthDto, dto: AssetStatsDto) {
     const stats = await this.assetRepository.getStatistics(auth.user.id, dto);
     return mapStats(stats);

server/src/services/audit.service.spec.ts

@@ -1,6 +1,4 @@
-import { BadRequestException } from '@nestjs/common';
-import { FileReportItemDto } from 'src/dtos/audit.dto';
-import { AssetFileType, AssetPathType, JobStatus, PersonPathType, UserPathType } from 'src/enum';
+import { JobStatus } from 'src/enum';
 import { AuditService } from 'src/services/audit.service';
 import { newTestService, ServiceMocks } from 'test/utils';
 
@@ -25,148 +23,4 @@ describe(AuditService.name, () => {
       expect(mocks.audit.removeBefore).toHaveBeenCalledWith(expect.any(Date));
     });
   });
-
-  describe('getChecksums', () => {
-    it('should fail if the file is not in the immich path', async () => {
-      await expect(sut.getChecksums({ filenames: ['foo/bar'] })).rejects.toBeInstanceOf(BadRequestException);
-
-      expect(mocks.crypto.hashFile).not.toHaveBeenCalled();
-    });
-
-    it('should get checksum for valid file', async () => {
-      await expect(sut.getChecksums({ filenames: ['./upload/my-file.jpg'] })).resolves.toEqual([
-        { filename: './upload/my-file.jpg', checksum: expect.any(String) },
-      ]);
-
-      expect(mocks.crypto.hashFile).toHaveBeenCalledWith('./upload/my-file.jpg');
-    });
-  });
-
-  describe('fixItems', () => {
-    it('should fail if the file is not in the immich path', async () => {
-      await expect(
-        sut.fixItems([
-          { entityId: 'my-id', pathType: AssetPathType.ORIGINAL, pathValue: 'foo/bar' } as FileReportItemDto,
-        ]),
-      ).rejects.toBeInstanceOf(BadRequestException);
-
-      expect(mocks.asset.update).not.toHaveBeenCalled();
-      expect(mocks.asset.upsertFile).not.toHaveBeenCalled();
-      expect(mocks.person.update).not.toHaveBeenCalled();
-      expect(mocks.user.update).not.toHaveBeenCalled();
-    });
-
-    it('should update encoded video path', async () => {
-      await sut.fixItems([
-        {
-          entityId: 'my-id',
-          pathType: AssetPathType.ENCODED_VIDEO,
-          pathValue: './upload/my-video.mp4',
-        } as FileReportItemDto,
-      ]);
-
-      expect(mocks.asset.update).toHaveBeenCalledWith({ id: 'my-id', encodedVideoPath: './upload/my-video.mp4' });
-      expect(mocks.asset.upsertFile).not.toHaveBeenCalled();
-      expect(mocks.person.update).not.toHaveBeenCalled();
-      expect(mocks.user.update).not.toHaveBeenCalled();
-    });
-
-    it('should update preview path', async () => {
-      await sut.fixItems([
-        {
-          entityId: 'my-id',
-          pathType: AssetPathType.PREVIEW,
-          pathValue: './upload/my-preview.png',
-        } as FileReportItemDto,
-      ]);
-
-      expect(mocks.asset.upsertFile).toHaveBeenCalledWith({
-        assetId: 'my-id',
-        type: AssetFileType.PREVIEW,
-        path: './upload/my-preview.png',
-      });
-      expect(mocks.asset.update).not.toHaveBeenCalled();
-      expect(mocks.person.update).not.toHaveBeenCalled();
-      expect(mocks.user.update).not.toHaveBeenCalled();
-    });
-
-    it('should update thumbnail path', async () => {
-      await sut.fixItems([
-        {
-          entityId: 'my-id',
-          pathType: AssetPathType.THUMBNAIL,
-          pathValue: './upload/my-thumbnail.webp',
-        } as FileReportItemDto,
-      ]);
-
-      expect(mocks.asset.upsertFile).toHaveBeenCalledWith({
-        assetId: 'my-id',
-        type: AssetFileType.THUMBNAIL,
-        path: './upload/my-thumbnail.webp',
-      });
-      expect(mocks.asset.update).not.toHaveBeenCalled();
-      expect(mocks.person.update).not.toHaveBeenCalled();
-      expect(mocks.user.update).not.toHaveBeenCalled();
-    });
-
-    it('should update original path', async () => {
-      await sut.fixItems([
-        {
-          entityId: 'my-id',
-          pathType: AssetPathType.ORIGINAL,
-          pathValue: './upload/my-original.png',
-        } as FileReportItemDto,
-      ]);
-
-      expect(mocks.asset.update).toHaveBeenCalledWith({ id: 'my-id', originalPath: './upload/my-original.png' });
-      expect(mocks.asset.upsertFile).not.toHaveBeenCalled();
-      expect(mocks.person.update).not.toHaveBeenCalled();
-      expect(mocks.user.update).not.toHaveBeenCalled();
-    });
-
-    it('should update sidecar path', async () => {
-      await sut.fixItems([
-        {
-          entityId: 'my-id',
-          pathType: AssetPathType.SIDECAR,
-          pathValue: './upload/my-sidecar.xmp',
-        } as FileReportItemDto,
-      ]);
-
-      expect(mocks.asset.update).toHaveBeenCalledWith({ id: 'my-id', sidecarPath: './upload/my-sidecar.xmp' });
-      expect(mocks.asset.upsertFile).not.toHaveBeenCalled();
-      expect(mocks.person.update).not.toHaveBeenCalled();
-      expect(mocks.user.update).not.toHaveBeenCalled();
-    });
-
-    it('should update face path', async () => {
-      await sut.fixItems([
-        {
-          entityId: 'my-id',
-          pathType: PersonPathType.FACE,
-          pathValue: './upload/my-face.jpg',
-        } as FileReportItemDto,
-      ]);
-
-      expect(mocks.person.update).toHaveBeenCalledWith({ id: 'my-id', thumbnailPath: './upload/my-face.jpg' });
-      expect(mocks.asset.update).not.toHaveBeenCalled();
-      expect(mocks.asset.upsertFile).not.toHaveBeenCalled();
-      expect(mocks.user.update).not.toHaveBeenCalled();
-    });
-
-    it('should update profile path', async () => {
-      await sut.fixItems([
-        {
-          entityId: 'my-id',
-          pathType: UserPathType.PROFILE,
-          pathValue: './upload/my-profile-pic.jpg',
-        } as FileReportItemDto,
-      ]);
-
-      expect(mocks.user.update).toHaveBeenCalledWith('my-id', { profileImagePath: './upload/my-profile-pic.jpg' });
-      expect(mocks.asset.update).not.toHaveBeenCalled();
-      expect(mocks.asset.upsertFile).not.toHaveBeenCalled();
-      expect(mocks.person.update).not.toHaveBeenCalled();
-    });
-  });
 });

server/src/services/audit.service.ts

@@ -1,23 +1,9 @@
-import { BadRequestException, Injectable } from '@nestjs/common';
+import { Injectable } from '@nestjs/common';
 import { DateTime } from 'luxon';
-import { resolve } from 'node:path';
-import { AUDIT_LOG_MAX_DURATION, JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
-import { StorageCore } from 'src/cores/storage.core';
+import { AUDIT_LOG_MAX_DURATION } from 'src/constants';
 import { OnJob } from 'src/decorators';
-import { FileChecksumDto, FileChecksumResponseDto, FileReportItemDto, PathEntityType } from 'src/dtos/audit.dto';
-import {
-  AssetFileType,
-  AssetPathType,
-  JobName,
-  JobStatus,
-  PersonPathType,
-  QueueName,
-  StorageFolder,
-  UserPathType,
-} from 'src/enum';
+import { JobName, JobStatus, QueueName } from 'src/enum';
 import { BaseService } from 'src/services/base.service';
-import { getAssetFiles } from 'src/utils/asset.util';
-import { usePagination } from 'src/utils/pagination';
 
 @Injectable()
 export class AuditService extends BaseService {
@@ -26,187 +12,4 @@ export class AuditService extends BaseService {
     await this.auditRepository.removeBefore(DateTime.now().minus(AUDIT_LOG_MAX_DURATION).toJSDate());
     return JobStatus.SUCCESS;
   }
-
-  async getChecksums(dto: FileChecksumDto) {
-    const results: FileChecksumResponseDto[] = [];
-    for (const filename of dto.filenames) {
-      if (!StorageCore.isImmichPath(filename)) {
-        throw new BadRequestException(
-          `Could not get the checksum of ${filename} because the file isn't accessible by Immich`,
-        );
-      }
-
-      const checksum = await this.cryptoRepository.hashFile(filename);
-      results.push({ filename, checksum: checksum.toString('base64') });
-    }
-    return results;
-  }
-
-  async fixItems(items: FileReportItemDto[]) {
-    for (const { entityId: id, pathType, pathValue } of items) {
-      if (!StorageCore.isImmichPath(pathValue)) {
-        throw new BadRequestException(
-          `Could not fix item ${id} with path ${pathValue} because the file isn't accessible by Immich`,
-        );
-      }
-
-      switch (pathType) {
-        case AssetPathType.ENCODED_VIDEO: {
-          await this.assetRepository.update({ id, encodedVideoPath: pathValue });
-          break;
-        }
-
-        case AssetPathType.PREVIEW: {
-          await this.assetRepository.upsertFile({ assetId: id, type: AssetFileType.PREVIEW, path: pathValue });
-          break;
-        }
-
-        case AssetPathType.THUMBNAIL: {
-          await this.assetRepository.upsertFile({ assetId: id, type: AssetFileType.THUMBNAIL, path: pathValue });
-          break;
-        }
-
-        case AssetPathType.ORIGINAL: {
-          await this.assetRepository.update({ id, originalPath: pathValue });
-          break;
-        }
-
-        case AssetPathType.SIDECAR: {
-          await this.assetRepository.update({ id, sidecarPath: pathValue });
-          break;
-        }
-
-        case PersonPathType.FACE: {
-          await this.personRepository.update({ id, thumbnailPath: pathValue });
-          break;
-        }
-
-        case UserPathType.PROFILE: {
-          await this.userRepository.update(id, { profileImagePath: pathValue });
-          break;
-        }
-      }
-    }
-  }
-
-  private fullPath(filename: string) {
-    return resolve(filename);
-  }
-
-  async getFileReport() {
-    const hasFile = (items: Set<string>, filename: string) => items.has(filename) || items.has(this.fullPath(filename));
-    const crawl = async (folder: StorageFolder) =>
-      new Set(
-        await this.storageRepository.crawl({
-          includeHidden: true,
-          pathsToCrawl: [StorageCore.getBaseFolder(folder)],
-        }),
-      );
-
-    const uploadFiles = await crawl(StorageFolder.UPLOAD);
-    const libraryFiles = await crawl(StorageFolder.LIBRARY);
-    const thumbFiles = await crawl(StorageFolder.THUMBNAILS);
-    const videoFiles = await crawl(StorageFolder.ENCODED_VIDEO);
-    const profileFiles = await crawl(StorageFolder.PROFILE);
-    const allFiles = new Set<string>();
-    for (const list of [libraryFiles, thumbFiles, videoFiles, profileFiles, uploadFiles]) {
-      for (const item of list) {
-        allFiles.add(item);
-      }
-    }
-
-    const track = (filename: string | null | undefined) => {
-      if (!filename) {
-        return;
-      }
-      allFiles.delete(filename);
-      allFiles.delete(this.fullPath(filename));
-    };
-
-    this.logger.log(
-      `Found ${libraryFiles.size} original files, ${thumbFiles.size} thumbnails, ${videoFiles.size} encoded videos, ${profileFiles.size} profile files`,
-    );
-    const pagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (options) =>
-      this.assetRepository.getAll(options, { withDeleted: true, withArchived: true }),
-    );
-
-    let assetCount = 0;
-
-    const orphans: FileReportItemDto[] = [];
-    for await (const assets of pagination) {
-      assetCount += assets.length;
-      for (const { id, files, originalPath, encodedVideoPath, isExternal, checksum } of assets) {
-        const { fullsizeFile, previewFile, thumbnailFile } = getAssetFiles(files);
-        for (const file of [
-          originalPath,
-          fullsizeFile?.path,
-          previewFile?.path,
-          encodedVideoPath,
-          thumbnailFile?.path,
-        ]) {
-          track(file);
-        }
-
-        const entity = { entityId: id, entityType: PathEntityType.ASSET, checksum: checksum.toString('base64') };
-        if (
-          originalPath &&
-          !hasFile(libraryFiles, originalPath) &&
-          !hasFile(uploadFiles, originalPath) &&
-          // Android motion assets
-          !hasFile(videoFiles, originalPath) &&
-          // ignore external library assets
-          !isExternal
-        ) {
-          orphans.push({ ...entity, pathType: AssetPathType.ORIGINAL, pathValue: originalPath });
-        }
-        if (previewFile && !hasFile(thumbFiles, previewFile.path)) {
-          orphans.push({ ...entity, pathType: AssetPathType.PREVIEW, pathValue: previewFile.path });
-        }
-        if (thumbnailFile && !hasFile(thumbFiles, thumbnailFile.path)) {
-          orphans.push({ ...entity, pathType: AssetPathType.THUMBNAIL, pathValue: thumbnailFile.path });
-        }
-        if (encodedVideoPath && !hasFile(videoFiles, encodedVideoPath)) {
-          orphans.push({ ...entity, pathType: AssetPathType.THUMBNAIL, pathValue: encodedVideoPath });
-        }
-      }
-    }
-
-    const users = await this.userRepository.getList();
-    for (const { id, profileImagePath } of users) {
-      track(profileImagePath);
-
-      const entity = { entityId: id, entityType: PathEntityType.USER };
-      if (profileImagePath && !hasFile(profileFiles, profileImagePath)) {
-        orphans.push({ ...entity, pathType: UserPathType.PROFILE, pathValue: profileImagePath });
-      }
-    }
-
-    let peopleCount = 0;
-    for await (const { id, thumbnailPath } of this.personRepository.getAll()) {
-      track(thumbnailPath);
-      const entity = { entityId: id, entityType: PathEntityType.PERSON };
-      if (thumbnailPath && !hasFile(thumbFiles, thumbnailPath)) {
-        orphans.push({ ...entity, pathType: PersonPathType.FACE, pathValue: thumbnailPath });
-      }
-
-      if (peopleCount === JOBS_ASSET_PAGINATION_SIZE) {
-        this.logger.log(`Found ${assetCount} assets, ${users.length} users, ${peopleCount} people`);
-        peopleCount = 0;
-      }
-    }
-
-    this.logger.log(`Found ${assetCount} assets, ${users.length} users, ${peopleCount} people`);
-
-    const extras: string[] = [];
-    for (const file of allFiles) {
-      extras.push(file);
-    }
-
-    // send as absolute paths
-    for (const orphan of orphans) {
-      orphan.pathValue = this.fullPath(orphan.pathValue);
-    }
-
-    return { orphans, extras };
-  }
 }

server/src/services/duplicate.service.spec.ts

@@ -1,10 +1,9 @@
 import { AssetFileType, AssetType, JobName, JobStatus } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { DuplicateService } from 'src/services/duplicate.service';
 import { SearchService } from 'src/services/search.service';
 import { assetStub } from 'test/fixtures/asset.stub';
 import { authStub } from 'test/fixtures/auth.stub';
-import { newTestService, ServiceMocks } from 'test/utils';
+import { makeStream, newTestService, ServiceMocks } from 'test/utils';
 import { beforeEach, vitest } from 'vitest';
 
 vitest.useFakeTimers();
@@ -113,14 +112,11 @@ describe(SearchService.name, () => {
     });
 
     it('should queue missing assets', async () => {
-      mocks.asset.getWithout.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForSearchDuplicates.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueSearchDuplicates({});
 
-      expect(mocks.asset.getWithout).toHaveBeenCalledWith({ skip: 0, take: 1000 }, WithoutProperty.DUPLICATE);
+      expect(mocks.assetJob.streamForSearchDuplicates).toHaveBeenCalledWith(undefined);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.DUPLICATE_DETECTION,
@@ -130,14 +126,11 @@ describe(SearchService.name, () => {
     });
 
     it('should queue all assets', async () => {
-      mocks.asset.getAll.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForSearchDuplicates.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueSearchDuplicates({ force: true });
 
-      expect(mocks.asset.getAll).toHaveBeenCalled();
+      expect(mocks.assetJob.streamForSearchDuplicates).toHaveBeenCalledWith(true);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.DUPLICATE_DETECTION,

server/src/services/duplicate.service.ts

@@ -5,13 +5,11 @@ import { mapAsset } from 'src/dtos/asset-response.dto';
 import { AuthDto } from 'src/dtos/auth.dto';
 import { DuplicateResponseDto } from 'src/dtos/duplicate.dto';
 import { AssetFileType, JobName, JobStatus, QueueName } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { AssetDuplicateResult } from 'src/repositories/search.repository';
 import { BaseService } from 'src/services/base.service';
-import { JobOf } from 'src/types';
+import { JobItem, JobOf } from 'src/types';
 import { getAssetFile } from 'src/utils/asset.util';
 import { isDuplicateDetectionEnabled } from 'src/utils/misc';
-import { usePagination } from 'src/utils/pagination';
 
 @Injectable()
 export class DuplicateService extends BaseService {
@@ -30,18 +28,22 @@ export class DuplicateService extends BaseService {
       return JobStatus.SKIPPED;
     }
 
-    const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) => {
-      return force
-        ? this.assetRepository.getAll(pagination, { isVisible: true })
-        : this.assetRepository.getWithout(pagination, WithoutProperty.DUPLICATE);
-    });
+    let jobs: JobItem[] = [];
+    const queueAll = async () => {
+      await this.jobRepository.queueAll(jobs);
+      jobs = [];
+    };
 
-    for await (const assets of assetPagination) {
-      await this.jobRepository.queueAll(
-        assets.map((asset) => ({ name: JobName.DUPLICATE_DETECTION, data: { id: asset.id } })),
-      );
+    const assets = this.assetJobRepository.streamForSearchDuplicates(force);
+    for await (const asset of assets) {
+      jobs.push({ name: JobName.DUPLICATE_DETECTION, data: { id: asset.id } });
+      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
+        await queueAll();
+      }
     }
 
+    await queueAll();
+
     return JobStatus.SUCCESS;
   }
 

server/src/services/job.service.spec.ts

@@ -239,10 +239,6 @@ describe(JobService.name, () => {
         item: { name: JobName.SIDECAR_DISCOVERY, data: { id: 'asset-1' } },
         jobs: [JobName.METADATA_EXTRACTION],
       },
-      {
-        item: { name: JobName.METADATA_EXTRACTION, data: { id: 'asset-1' } },
-        jobs: [JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE],
-      },
       {
         item: { name: JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE, data: { id: 'asset-1', source: 'upload' } },
         jobs: [JobName.GENERATE_THUMBNAILS],

server/src/services/job.service.ts

@@ -264,17 +264,6 @@ export class JobService extends BaseService {
         break;
       }
 
-      case JobName.METADATA_EXTRACTION: {
-        if (item.data.source === 'sidecar-write') {
-          const [asset] = await this.assetRepository.getByIdsWithAllRelationsButStacks([item.data.id]);
-          if (asset) {
-            this.eventRepository.clientSend('on_asset_update', asset.ownerId, mapAsset(asset));
-          }
-        }
-        await this.jobRepository.queue({ name: JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE, data: item.data });
-        break;
-      }
-
       case JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE: {
         if (item.data.source === 'upload' || item.data.source === 'copy') {
           await this.jobRepository.queue({ name: JobName.GENERATE_THUMBNAILS, data: item.data });

server/src/services/library.service.spec.ts

@@ -273,7 +273,6 @@ describe(LibraryService.name, () => {
 
       mocks.library.get.mockResolvedValue(library);
       mocks.storage.walk.mockImplementation(async function* generator() {});
-      mocks.asset.getAll.mockResolvedValue({ items: [assetStub.external], hasNextPage: false });
       mocks.asset.getLibraryAssetCount.mockResolvedValue(1);
       mocks.asset.detectOfflineExternalAssets.mockResolvedValue({ numUpdatedRows: BigInt(1) });
 
@@ -292,7 +291,6 @@ describe(LibraryService.name, () => {
 
       mocks.library.get.mockResolvedValue(library);
       mocks.storage.walk.mockImplementation(async function* generator() {});
-      mocks.asset.getAll.mockResolvedValue({ items: [assetStub.external], hasNextPage: false });
       mocks.asset.getLibraryAssetCount.mockResolvedValue(0);
       mocks.asset.detectOfflineExternalAssets.mockResolvedValue({ numUpdatedRows: BigInt(1) });
 

server/src/services/media.service.spec.ts

@@ -38,10 +38,6 @@ describe(MediaService.name, () => {
   describe('handleQueueGenerateThumbnails', () => {
     it('should queue all assets', async () => {
       mocks.assetJob.streamForThumbnailJob.mockReturnValue(makeStream([assetStub.image]));
-      mocks.asset.getAll.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
 
       mocks.person.getAll.mockReturnValue(makeStream([personStub.newThumbnail]));
       mocks.person.getFacesByIds.mockResolvedValue([faceStub.face1]);
@@ -67,10 +63,6 @@ describe(MediaService.name, () => {
 
     it('should queue trashed assets when force is true', async () => {
       mocks.assetJob.streamForThumbnailJob.mockReturnValue(makeStream([assetStub.archived]));
-      mocks.asset.getAll.mockResolvedValue({
-        items: [assetStub.trashed],
-        hasNextPage: false,
-      });
       mocks.person.getAll.mockReturnValue(makeStream());
 
       await sut.handleQueueGenerateThumbnails({ force: true });
@@ -171,7 +163,7 @@ describe(MediaService.name, () => {
 
   describe('handleQueueMigration', () => {
     it('should remove empty directories and queue jobs', async () => {
-      mocks.asset.getAll.mockResolvedValue({ hasNextPage: false, items: [assetStub.image] });
+      mocks.assetJob.streamForMigrationJob.mockReturnValue(makeStream([assetStub.image]));
       mocks.job.getJobCounts.mockResolvedValue({ active: 1, waiting: 0 } as JobCounts);
       mocks.person.getAll.mockReturnValue(makeStream([personStub.withName]));
 

server/src/services/media.service.ts

@@ -36,7 +36,6 @@ import {
 import { getAssetFiles } from 'src/utils/asset.util';
 import { BaseConfig, ThumbnailConfig } from 'src/utils/media';
 import { mimeTypes } from 'src/utils/mime-types';
-import { usePagination } from 'src/utils/pagination';
 
 @Injectable()
 export class MediaService extends BaseService {
@@ -50,18 +49,26 @@ export class MediaService extends BaseService {
 
   @OnJob({ name: JobName.QUEUE_GENERATE_THUMBNAILS, queue: QueueName.THUMBNAIL_GENERATION })
   async handleQueueGenerateThumbnails({ force }: JobOf<JobName.QUEUE_GENERATE_THUMBNAILS>): Promise<JobStatus> {
-    const thumbJobs: JobItem[] = [];
+    let jobs: JobItem[] = [];
+
+    const queueAll = async () => {
+      await this.jobRepository.queueAll(jobs);
+      jobs = [];
+    };
+
     for await (const asset of this.assetJobRepository.streamForThumbnailJob(!!force)) {
       const { previewFile, thumbnailFile } = getAssetFiles(asset.files);
 
       if (!previewFile || !thumbnailFile || !asset.thumbhash || force) {
-        thumbJobs.push({ name: JobName.GENERATE_THUMBNAILS, data: { id: asset.id } });
-        continue;
+        jobs.push({ name: JobName.GENERATE_THUMBNAILS, data: { id: asset.id } });
+      }
+
+      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
+        await queueAll();
       }
     }
-    await this.jobRepository.queueAll(thumbJobs);
 
-    const jobs: JobItem[] = [];
+    await queueAll();
 
     const people = this.personRepository.getAll(force ? undefined : { thumbnailPath: '' });
 
@@ -76,32 +83,36 @@ export class MediaService extends BaseService {
       }
 
       jobs.push({ name: JobName.GENERATE_PERSON_THUMBNAIL, data: { id: person.id } });
+      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
+        await queueAll();
+      }
     }
 
-    await this.jobRepository.queueAll(jobs);
+    await queueAll();
 
     return JobStatus.SUCCESS;
   }
 
   @OnJob({ name: JobName.QUEUE_MIGRATION, queue: QueueName.MIGRATION })
   async handleQueueMigration(): Promise<JobStatus> {
-    const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) =>
-      this.assetRepository.getAll(pagination),
-    );
-
     const { active, waiting } = await this.jobRepository.getJobCounts(QueueName.MIGRATION);
     if (active === 1 && waiting === 0) {
       await this.storageCore.removeEmptyDirs(StorageFolder.THUMBNAILS);
       await this.storageCore.removeEmptyDirs(StorageFolder.ENCODED_VIDEO);
     }
 
-    for await (const assets of assetPagination) {
-      await this.jobRepository.queueAll(
-        assets.map((asset) => ({ name: JobName.MIGRATE_ASSET, data: { id: asset.id } })),
-      );
+    let jobs: JobItem[] = [];
+    const assets = this.assetJobRepository.streamForMigrationJob();
+    for await (const asset of assets) {
+      jobs.push({ name: JobName.MIGRATE_ASSET, data: { id: asset.id } });
+      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
+        await this.jobRepository.queueAll(jobs);
+        jobs = [];
+      }
     }
 
-    let jobs: { name: JobName.MIGRATE_PERSON; data: { id: string } }[] = [];
+    await this.jobRepository.queueAll(jobs);
+    jobs = [];
 
     for await (const person of this.personRepository.getAll()) {
       jobs.push({ name: JobName.MIGRATE_PERSON, data: { id: person.id } });
@@ -255,7 +266,9 @@ export class MediaService extends BaseService {
 
     const { info, data, colorspace } = await this.decodeImage(
       extracted ? extracted.buffer : asset.originalPath,
-      asset.exifInfo,
+      // only specify orientation to extracted images which don't have EXIF orientation data
+      // or it can double rotate the image
+      extracted ? asset.exifInfo : { ...asset.exifInfo, orientation: null },
       convertFullsize ? undefined : image.preview.size,
     );
 

server/src/services/metadata.service.spec.ts

@@ -5,7 +5,6 @@ import { constants } from 'node:fs/promises';
 import { defaults } from 'src/config';
 import { MapAsset } from 'src/dtos/asset-response.dto';
 import { AssetType, ExifOrientation, ImmichWorker, JobName, JobStatus, SourceType } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { ImmichTags } from 'src/repositories/metadata.repository';
 import { MetadataService } from 'src/services/metadata.service';
 import { assetStub } from 'test/fixtures/asset.stub';
@@ -144,7 +143,8 @@ describe(MetadataService.name, () => {
 
     it('should handle an asset that could not be found', async () => {
       mocks.assetJob.getForMetadataExtraction.mockResolvedValue(void 0);
-      await expect(sut.handleMetadataExtraction({ id: assetStub.image.id })).resolves.toBe(JobStatus.FAILED);
+
+      await sut.handleMetadataExtraction({ id: assetStub.image.id });
 
       expect(mocks.assetJob.getForMetadataExtraction).toHaveBeenCalledWith(assetStub.image.id);
       expect(mocks.asset.upsertExif).not.toHaveBeenCalled();
@@ -527,7 +527,7 @@ describe(MetadataService.name, () => {
         ContainerDirectory: [{ Foo: 100 }],
       });
 
-      await expect(sut.handleMetadataExtraction({ id: assetStub.image.id })).resolves.toBe(JobStatus.SUCCESS);
+      await sut.handleMetadataExtraction({ id: assetStub.image.id });
     });
 
     it('should extract the correct video orientation', async () => {
@@ -1202,7 +1202,7 @@ describe(MetadataService.name, () => {
     it('should handle livePhotoCID not set', async () => {
       mocks.assetJob.getForMetadataExtraction.mockResolvedValue(assetStub.image);
 
-      await expect(sut.handleMetadataExtraction({ id: assetStub.image.id })).resolves.toBe(JobStatus.SUCCESS);
+      await sut.handleMetadataExtraction({ id: assetStub.image.id });
 
       expect(mocks.assetJob.getForMetadataExtraction).toHaveBeenCalledWith(assetStub.image.id);
       expect(mocks.asset.findLivePhotoMatch).not.toHaveBeenCalled();
@@ -1215,9 +1215,7 @@ describe(MetadataService.name, () => {
       mocks.assetJob.getForMetadataExtraction.mockResolvedValue(assetStub.livePhotoMotionAsset);
       mockReadTags({ ContentIdentifier: 'CID' });
 
-      await expect(sut.handleMetadataExtraction({ id: assetStub.livePhotoMotionAsset.id })).resolves.toBe(
-        JobStatus.SUCCESS,
-      );
+      await sut.handleMetadataExtraction({ id: assetStub.livePhotoMotionAsset.id });
 
       expect(mocks.assetJob.getForMetadataExtraction).toHaveBeenCalledWith(assetStub.livePhotoMotionAsset.id);
       expect(mocks.asset.findLivePhotoMatch).toHaveBeenCalledWith({
@@ -1236,9 +1234,7 @@ describe(MetadataService.name, () => {
       mocks.asset.findLivePhotoMatch.mockResolvedValue(assetStub.livePhotoMotionAsset);
       mockReadTags({ ContentIdentifier: 'CID' });
 
-      await expect(sut.handleMetadataExtraction({ id: assetStub.livePhotoStillAsset.id })).resolves.toBe(
-        JobStatus.SUCCESS,
-      );
+      await sut.handleMetadataExtraction({ id: assetStub.livePhotoStillAsset.id });
 
       expect(mocks.assetJob.getForMetadataExtraction).toHaveBeenCalledWith(assetStub.livePhotoStillAsset.id);
       expect(mocks.asset.findLivePhotoMatch).toHaveBeenCalledWith({
@@ -1262,9 +1258,7 @@ describe(MetadataService.name, () => {
       mocks.asset.findLivePhotoMatch.mockResolvedValue(assetStub.livePhotoMotionAsset);
       mockReadTags({ ContentIdentifier: 'CID' });
 
-      await expect(sut.handleMetadataExtraction({ id: assetStub.livePhotoStillAsset.id })).resolves.toBe(
-        JobStatus.SUCCESS,
-      );
+      await sut.handleMetadataExtraction({ id: assetStub.livePhotoStillAsset.id });
 
       expect(mocks.event.emit).toHaveBeenCalledWith('asset.hide', {
         userId: assetStub.livePhotoMotionAsset.ownerId,
@@ -1280,10 +1274,12 @@ describe(MetadataService.name, () => {
       mocks.asset.findLivePhotoMatch.mockResolvedValue(assetStub.livePhotoMotionAsset);
       mockReadTags({ ContentIdentifier: 'CID' });
 
-      await expect(sut.handleMetadataExtraction({ id: assetStub.livePhotoStillAsset.id })).resolves.toBe(
-        JobStatus.SUCCESS,
-      );
+      await sut.handleMetadataExtraction({ id: assetStub.livePhotoStillAsset.id });
 
+      expect(mocks.event.emit).toHaveBeenCalledWith('asset.metadataExtracted', {
+        assetId: assetStub.livePhotoStillAsset.id,
+        userId: assetStub.livePhotoStillAsset.ownerId,
+      });
       expect(mocks.asset.findLivePhotoMatch).toHaveBeenCalledWith({
         ownerId: 'user-id',
         otherAssetId: 'live-photo-still-asset',
@@ -1346,12 +1342,11 @@ describe(MetadataService.name, () => {
 
   describe('handleQueueSidecar', () => {
     it('should queue assets with sidecar files', async () => {
-      mocks.asset.getAll.mockResolvedValue({ items: [assetStub.sidecar], hasNextPage: false });
+      mocks.assetJob.streamForSidecar.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueSidecar({ force: true });
+      expect(mocks.assetJob.streamForSidecar).toHaveBeenCalledWith(true);
 
-      expect(mocks.asset.getAll).toHaveBeenCalledWith({ take: 1000, skip: 0 });
-      expect(mocks.asset.getWithout).not.toHaveBeenCalled();
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.SIDECAR_SYNC,
@@ -1361,12 +1356,11 @@ describe(MetadataService.name, () => {
     });
 
     it('should queue assets without sidecar files', async () => {
-      mocks.asset.getWithout.mockResolvedValue({ items: [assetStub.image], hasNextPage: false });
+      mocks.assetJob.streamForSidecar.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueSidecar({ force: false });
 
-      expect(mocks.asset.getWithout).toHaveBeenCalledWith({ take: 1000, skip: 0 }, WithoutProperty.SIDECAR);
-      expect(mocks.asset.getAll).not.toHaveBeenCalled();
+      expect(mocks.assetJob.streamForSidecar).toHaveBeenCalledWith(false);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.SIDECAR_DISCOVERY,

server/src/services/metadata.service.ts

@@ -22,14 +22,12 @@ import {
   QueueName,
   SourceType,
 } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { ArgOf } from 'src/repositories/event.repository';
 import { ReverseGeocodeResult } from 'src/repositories/map.repository';
 import { ImmichTags } from 'src/repositories/metadata.repository';
 import { BaseService } from 'src/services/base.service';
-import { JobOf } from 'src/types';
+import { JobItem, JobOf } from 'src/types';
 import { isFaceImportEnabled } from 'src/utils/misc';
-import { usePagination } from 'src/utils/pagination';
 import { upsertTags } from 'src/utils/tag';
 
 /** look for a date from these tags (in order) */
@@ -184,14 +182,14 @@ export class MetadataService extends BaseService {
   }
 
   @OnJob({ name: JobName.METADATA_EXTRACTION, queue: QueueName.METADATA_EXTRACTION })
-  async handleMetadataExtraction(data: JobOf<JobName.METADATA_EXTRACTION>): Promise<JobStatus> {
+  async handleMetadataExtraction(data: JobOf<JobName.METADATA_EXTRACTION>) {
     const [{ metadata, reverseGeocoding }, asset] = await Promise.all([
       this.getConfig({ withCache: true }),
       this.assetJobRepository.getForMetadataExtraction(data.id),
     ]);
 
     if (!asset) {
-      return JobStatus.FAILED;
+      return;
     }
 
     const [exifTags, stats] = await Promise.all([
@@ -285,27 +283,31 @@ export class MetadataService extends BaseService {
 
     await this.assetRepository.upsertJobStatus({ assetId: asset.id, metadataExtractedAt: new Date() });
 
-    return JobStatus.SUCCESS;
+    await this.eventRepository.emit('asset.metadataExtracted', {
+      assetId: asset.id,
+      userId: asset.ownerId,
+      source: data.source,
+    });
   }
 
   @OnJob({ name: JobName.QUEUE_SIDECAR, queue: QueueName.SIDECAR })
-  async handleQueueSidecar(job: JobOf<JobName.QUEUE_SIDECAR>): Promise<JobStatus> {
-    const { force } = job;
-    const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) => {
-      return force
-        ? this.assetRepository.getAll(pagination)
-        : this.assetRepository.getWithout(pagination, WithoutProperty.SIDECAR);
-    });
+  async handleQueueSidecar({ force }: JobOf<JobName.QUEUE_SIDECAR>): Promise<JobStatus> {
+    let jobs: JobItem[] = [];
+    const queueAll = async () => {
+      await this.jobRepository.queueAll(jobs);
+      jobs = [];
+    };
 
-    for await (const assets of assetPagination) {
-      await this.jobRepository.queueAll(
-        assets.map((asset) => ({
-          name: force ? JobName.SIDECAR_SYNC : JobName.SIDECAR_DISCOVERY,
-          data: { id: asset.id },
-        })),
-      );
+    const assets = this.assetJobRepository.streamForSidecar(force);
+    for await (const asset of assets) {
+      jobs.push({ name: force ? JobName.SIDECAR_SYNC : JobName.SIDECAR_DISCOVERY, data: { id: asset.id } });
+      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
+        await queueAll();
+      }
     }
 
+    await queueAll();
+
     return JobStatus.SUCCESS;
   }
 

server/src/services/notification.service.spec.ts

@@ -154,10 +154,10 @@ describe(NotificationService.name, () => {
 
   describe('onAlbumUpdateEvent', () => {
     it('should queue notify album update event', async () => {
-      await sut.onAlbumUpdate({ id: 'album', recipientIds: ['42'] });
+      await sut.onAlbumUpdate({ id: 'album', recipientId: '42' });
       expect(mocks.job.queue).toHaveBeenCalledWith({
         name: JobName.NOTIFY_ALBUM_UPDATE,
-        data: { id: 'album', recipientIds: ['42'], delay: 300_000 },
+        data: { id: 'album', recipientId: '42', delay: 300_000 },
       });
     });
   });
@@ -414,14 +414,14 @@ describe(NotificationService.name, () => {
 
   describe('handleAlbumUpdate', () => {
     it('should skip if album could not be found', async () => {
-      await expect(sut.handleAlbumUpdate({ id: '', recipientIds: ['1'] })).resolves.toBe(JobStatus.SKIPPED);
+      await expect(sut.handleAlbumUpdate({ id: '', recipientId: '1' })).resolves.toBe(JobStatus.SKIPPED);
       expect(mocks.user.get).not.toHaveBeenCalled();
     });
 
     it('should skip if owner could not be found', async () => {
       mocks.album.getById.mockResolvedValue(albumStub.emptyWithValidThumbnail);
 
-      await expect(sut.handleAlbumUpdate({ id: '', recipientIds: ['1'] })).resolves.toBe(JobStatus.SKIPPED);
+      await expect(sut.handleAlbumUpdate({ id: '', recipientId: '1' })).resolves.toBe(JobStatus.SKIPPED);
       expect(mocks.systemMetadata.get).not.toHaveBeenCalled();
     });
 
@@ -434,7 +434,7 @@ describe(NotificationService.name, () => {
       mocks.email.renderEmail.mockResolvedValue({ html: '', text: '' });
       mocks.assetJob.getAlbumThumbnailFiles.mockResolvedValue([]);
 
-      await sut.handleAlbumUpdate({ id: '', recipientIds: [userStub.user1.id] });
+      await sut.handleAlbumUpdate({ id: '', recipientId: userStub.user1.id });
       expect(mocks.user.get).toHaveBeenCalledWith(userStub.user1.id, { withDeleted: false });
       expect(mocks.email.renderEmail).not.toHaveBeenCalled();
     });
@@ -456,7 +456,7 @@ describe(NotificationService.name, () => {
       mocks.email.renderEmail.mockResolvedValue({ html: '', text: '' });
       mocks.assetJob.getAlbumThumbnailFiles.mockResolvedValue([]);
 
-      await sut.handleAlbumUpdate({ id: '', recipientIds: [userStub.user1.id] });
+      await sut.handleAlbumUpdate({ id: '', recipientId: userStub.user1.id });
       expect(mocks.user.get).toHaveBeenCalledWith(userStub.user1.id, { withDeleted: false });
       expect(mocks.email.renderEmail).not.toHaveBeenCalled();
     });
@@ -478,7 +478,7 @@ describe(NotificationService.name, () => {
       mocks.email.renderEmail.mockResolvedValue({ html: '', text: '' });
       mocks.assetJob.getAlbumThumbnailFiles.mockResolvedValue([]);
 
-      await sut.handleAlbumUpdate({ id: '', recipientIds: [userStub.user1.id] });
+      await sut.handleAlbumUpdate({ id: '', recipientId: userStub.user1.id });
       expect(mocks.user.get).toHaveBeenCalledWith(userStub.user1.id, { withDeleted: false });
       expect(mocks.email.renderEmail).not.toHaveBeenCalled();
     });
@@ -492,21 +492,21 @@ describe(NotificationService.name, () => {
       mocks.email.renderEmail.mockResolvedValue({ html: '', text: '' });
       mocks.assetJob.getAlbumThumbnailFiles.mockResolvedValue([]);
 
-      await sut.handleAlbumUpdate({ id: '', recipientIds: [userStub.user1.id] });
+      await sut.handleAlbumUpdate({ id: '', recipientId: userStub.user1.id });
       expect(mocks.user.get).toHaveBeenCalledWith(userStub.user1.id, { withDeleted: false });
       expect(mocks.email.renderEmail).toHaveBeenCalled();
       expect(mocks.job.queue).toHaveBeenCalled();
     });
 
     it('should add new recipients for new images if job is already queued', async () => {
-      mocks.job.removeJob.mockResolvedValue({ id: '1', recipientIds: ['2', '3', '4'] } as INotifyAlbumUpdateJob);
-      await sut.onAlbumUpdate({ id: '1', recipientIds: ['1', '2', '3'] } as INotifyAlbumUpdateJob);
+      await sut.onAlbumUpdate({ id: '1', recipientId: '2' } as INotifyAlbumUpdateJob);
+      expect(mocks.job.removeJob).toHaveBeenCalledWith(JobName.NOTIFY_ALBUM_UPDATE, '1/2');
       expect(mocks.job.queue).toHaveBeenCalledWith({
         name: JobName.NOTIFY_ALBUM_UPDATE,
         data: {
           id: '1',
           delay: 300_000,
-          recipientIds: ['1', '2', '3', '4'],
+          recipientId: '2',
         },
       });
     });

server/src/services/notification.service.ts

@@ -1,5 +1,6 @@
 import { BadRequestException, Injectable } from '@nestjs/common';
 import { OnEvent, OnJob } from 'src/decorators';
+import { mapAsset } from 'src/dtos/asset-response.dto';
 import { AuthDto } from 'src/dtos/auth.dto';
 import {
   mapNotification,
@@ -22,7 +23,7 @@ import {
 import { EmailTemplate } from 'src/repositories/email.repository';
 import { ArgOf } from 'src/repositories/event.repository';
 import { BaseService } from 'src/services/base.service';
-import { EmailImageAttachment, IEntityJob, INotifyAlbumUpdateJob, JobItem, JobOf } from 'src/types';
+import { EmailImageAttachment, JobOf } from 'src/types';
 import { getFilenameExtension } from 'src/utils/file';
 import { getExternalDomain } from 'src/utils/misc';
 import { isEqualObject } from 'src/utils/object';
@@ -152,6 +153,18 @@ export class NotificationService extends BaseService {
     this.eventRepository.clientSend('on_asset_trash', userId, assetIds);
   }
 
+  @OnEvent({ name: 'asset.metadataExtracted' })
+  async onAssetMetadataExtracted({ assetId, userId, source }: ArgOf<'asset.metadataExtracted'>) {
+    if (source !== 'sidecar-write') {
+      return;
+    }
+
+    const [asset] = await this.assetRepository.getByIdsWithAllRelationsButStacks([assetId]);
+    if (asset) {
+      this.eventRepository.clientSend('on_asset_update', userId, mapAsset(asset));
+    }
+  }
+
   @OnEvent({ name: 'assets.restore' })
   onAssetsRestore({ assetIds, userId }: ArgOf<'assets.restore'>) {
     this.eventRepository.clientSend('on_asset_restore', userId, assetIds);
@@ -185,30 +198,12 @@ export class NotificationService extends BaseService {
   }
 
   @OnEvent({ name: 'album.update' })
-  async onAlbumUpdate({ id, recipientIds }: ArgOf<'album.update'>) {
-    // if recipientIds is empty, album likely only has one user part of it, don't queue notification if so
-    if (recipientIds.length === 0) {
-      return;
-    }
-
-    const job: JobItem = {
+  async onAlbumUpdate({ id, recipientId }: ArgOf<'album.update'>) {
+    await this.jobRepository.removeJob(JobName.NOTIFY_ALBUM_UPDATE, `${id}/${recipientId}`);
+    await this.jobRepository.queue({
       name: JobName.NOTIFY_ALBUM_UPDATE,
-      data: { id, recipientIds, delay: NotificationService.albumUpdateEmailDelayMs },
-    };
-
-    const previousJobData = await this.jobRepository.removeJob(id, JobName.NOTIFY_ALBUM_UPDATE);
-    if (previousJobData && this.isAlbumUpdateJob(previousJobData)) {
-      for (const id of previousJobData.recipientIds) {
-        if (!recipientIds.includes(id)) {
-          recipientIds.push(id);
-        }
-      }
-    }
-    await this.jobRepository.queue(job);
-  }
-
-  private isAlbumUpdateJob(job: IEntityJob): job is INotifyAlbumUpdateJob {
-    return 'recipientIds' in job;
+      data: { id, recipientId, delay: NotificationService.albumUpdateEmailDelayMs },
+    });
   }
 
   @OnEvent({ name: 'album.invite' })
@@ -399,7 +394,7 @@ export class NotificationService extends BaseService {
   }
 
   @OnJob({ name: JobName.NOTIFY_ALBUM_UPDATE, queue: QueueName.NOTIFICATION })
-  async handleAlbumUpdate({ id, recipientIds }: JobOf<JobName.NOTIFY_ALBUM_UPDATE>) {
+  async handleAlbumUpdate({ id, recipientId }: JobOf<JobName.NOTIFY_ALBUM_UPDATE>) {
     const album = await this.albumRepository.getById(id, { withAssets: false });
 
     if (!album) {
@@ -411,49 +406,44 @@ export class NotificationService extends BaseService {
       return JobStatus.SKIPPED;
     }
 
-    const recipients = [...album.albumUsers.map((user) => user.user), owner].filter((user) =>
-      recipientIds.includes(user.id),
-    );
     const attachment = await this.getAlbumThumbnailAttachment(album);
 
     const { server, templates } = await this.getConfig({ withCache: false });
 
-    for (const recipient of recipients) {
-      const user = await this.userRepository.get(recipient.id, { withDeleted: false });
-      if (!user) {
-        continue;
-      }
-
-      const { emailNotifications } = getPreferences(user.metadata);
-
-      if (!emailNotifications.enabled || !emailNotifications.albumUpdate) {
-        continue;
-      }
-
-      const { html, text } = await this.emailRepository.renderEmail({
-        template: EmailTemplate.ALBUM_UPDATE,
-        data: {
-          baseUrl: getExternalDomain(server),
-          albumId: album.id,
-          albumName: album.albumName,
-          recipientName: recipient.name,
-          cid: attachment ? attachment.cid : undefined,
-        },
-        customTemplate: templates.email.albumUpdateTemplate,
-      });
-
-      await this.jobRepository.queue({
-        name: JobName.SEND_EMAIL,
-        data: {
-          to: recipient.email,
-          subject: `New media has been added to an album - ${album.albumName}`,
-          html,
-          text,
-          imageAttachments: attachment ? [attachment] : undefined,
-        },
-      });
+    const user = await this.userRepository.get(recipientId, { withDeleted: false });
+    if (!user) {
+      return JobStatus.SKIPPED;
     }
 
+    const { emailNotifications } = getPreferences(user.metadata);
+
+    if (!emailNotifications.enabled || !emailNotifications.albumUpdate) {
+      return JobStatus.SKIPPED;
+    }
+
+    const { html, text } = await this.emailRepository.renderEmail({
+      template: EmailTemplate.ALBUM_UPDATE,
+      data: {
+        baseUrl: getExternalDomain(server),
+        albumId: album.id,
+        albumName: album.albumName,
+        recipientName: user.name,
+        cid: attachment ? attachment.cid : undefined,
+      },
+      customTemplate: templates.email.albumUpdateTemplate,
+    });
+
+    await this.jobRepository.queue({
+      name: JobName.SEND_EMAIL,
+      data: {
+        to: user.email,
+        subject: `New media has been added to an album - ${album.albumName}`,
+        html,
+        text,
+        imageAttachments: attachment ? [attachment] : undefined,
+      },
+    });
+
     return JobStatus.SUCCESS;
   }
 

server/src/services/person.service.spec.ts

@@ -2,7 +2,6 @@ import { BadRequestException, NotFoundException } from '@nestjs/common';
 import { BulkIdErrorReason } from 'src/dtos/asset-ids.response.dto';
 import { mapFaces, mapPerson, PersonResponseDto } from 'src/dtos/person.dto';
 import { CacheControl, Colorspace, ImageFormat, JobName, JobStatus, SourceType, SystemMetadataKey } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { DetectedFaces } from 'src/repositories/machine-learning.repository';
 import { FaceSearchResult } from 'src/repositories/search.repository';
 import { PersonService } from 'src/services/person.service';
@@ -455,14 +454,11 @@ describe(PersonService.name, () => {
     });
 
     it('should queue missing assets', async () => {
-      mocks.asset.getWithout.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForDetectFacesJob.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueDetectFaces({ force: false });
 
-      expect(mocks.asset.getWithout).toHaveBeenCalledWith({ skip: 0, take: 1000 }, WithoutProperty.FACES);
+      expect(mocks.assetJob.streamForDetectFacesJob).toHaveBeenCalledWith(false);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.FACE_DETECTION,
@@ -472,10 +468,7 @@ describe(PersonService.name, () => {
     });
 
     it('should queue all assets', async () => {
-      mocks.asset.getAll.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForDetectFacesJob.mockReturnValue(makeStream([assetStub.image]));
       mocks.person.getAllWithoutFaces.mockResolvedValue([personStub.withName]);
 
       await sut.handleQueueDetectFaces({ force: true });
@@ -483,7 +476,7 @@ describe(PersonService.name, () => {
       expect(mocks.person.deleteFaces).toHaveBeenCalledWith({ sourceType: SourceType.MACHINE_LEARNING });
       expect(mocks.person.delete).toHaveBeenCalledWith([personStub.withName.id]);
       expect(mocks.storage.unlink).toHaveBeenCalledWith(personStub.withName.thumbnailPath);
-      expect(mocks.asset.getAll).toHaveBeenCalled();
+      expect(mocks.assetJob.streamForDetectFacesJob).toHaveBeenCalledWith(true);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.FACE_DETECTION,
@@ -493,17 +486,14 @@ describe(PersonService.name, () => {
     });
 
     it('should refresh all assets', async () => {
-      mocks.asset.getAll.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForDetectFacesJob.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueDetectFaces({ force: undefined });
 
       expect(mocks.person.delete).not.toHaveBeenCalled();
       expect(mocks.person.deleteFaces).not.toHaveBeenCalled();
       expect(mocks.storage.unlink).not.toHaveBeenCalled();
-      expect(mocks.asset.getAll).toHaveBeenCalled();
+      expect(mocks.assetJob.streamForDetectFacesJob).toHaveBeenCalledWith(undefined);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.FACE_DETECTION,
@@ -516,16 +506,13 @@ describe(PersonService.name, () => {
     it('should delete existing people and faces if forced', async () => {
       mocks.person.getAll.mockReturnValue(makeStream([faceStub.face1.person, personStub.randomPerson]));
       mocks.person.getAllFaces.mockReturnValue(makeStream([faceStub.face1]));
-      mocks.asset.getAll.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForDetectFacesJob.mockReturnValue(makeStream([assetStub.image]));
       mocks.person.getAllWithoutFaces.mockResolvedValue([personStub.randomPerson]);
       mocks.person.deleteFaces.mockResolvedValue();
 
       await sut.handleQueueDetectFaces({ force: true });
 
-      expect(mocks.asset.getAll).toHaveBeenCalled();
+      expect(mocks.assetJob.streamForDetectFacesJob).toHaveBeenCalledWith(true);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.FACE_DETECTION,

server/src/services/person.service.ts

@@ -36,7 +36,6 @@ import {
   SourceType,
   SystemMetadataKey,
 } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { BoundingBox } from 'src/repositories/machine-learning.repository';
 import { UpdateFacesData } from 'src/repositories/person.repository';
 import { BaseService } from 'src/services/base.service';
@@ -44,7 +43,6 @@ import { CropOptions, ImageDimensions, InputDimensions, JobItem, JobOf } from 's
 import { ImmichFileResponse } from 'src/utils/file';
 import { mimeTypes } from 'src/utils/mime-types';
 import { isFaceImportEnabled, isFacialRecognitionEnabled } from 'src/utils/misc';
-import { usePagination } from 'src/utils/pagination';
 
 @Injectable()
 export class PersonService extends BaseService {
@@ -265,23 +263,19 @@ export class PersonService extends BaseService {
       await this.handlePersonCleanup();
     }
 
-    const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) => {
-      return force === false
-        ? this.assetRepository.getWithout(pagination, WithoutProperty.FACES)
-        : this.assetRepository.getAll(pagination, {
-            orderDirection: 'desc',
-            withFaces: true,
-            withArchived: true,
-            isVisible: true,
-          });
-    });
+    let jobs: JobItem[] = [];
+    const assets = this.assetJobRepository.streamForDetectFacesJob(force);
+    for await (const asset of assets) {
+      jobs.push({ name: JobName.FACE_DETECTION, data: { id: asset.id } });
 
-    for await (const assets of assetPagination) {
-      await this.jobRepository.queueAll(
-        assets.map((asset) => ({ name: JobName.FACE_DETECTION, data: { id: asset.id } })),
-      );
+      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
+        await this.jobRepository.queueAll(jobs);
+        jobs = [];
+      }
     }
 
+    await this.jobRepository.queueAll(jobs);
+
     if (force === undefined) {
       await this.jobRepository.queue({ name: JobName.PERSON_CLEANUP });
     }

server/src/services/search.service.ts

@@ -15,7 +15,6 @@ import {
   SmartSearchDto,
 } from 'src/dtos/search.dto';
 import { AssetOrder } from 'src/enum';
-import { SearchExploreItem } from 'src/repositories/search.repository';
 import { BaseService } from 'src/services/base.service';
 import { getMyPartnerIds } from 'src/utils/asset.util';
 import { isSmartSearchEnabled } from 'src/utils/misc';
@@ -32,7 +31,7 @@ export class SearchService extends BaseService {
     return places.map((place) => mapPlaces(place));
   }
 
-  async getExploreData(auth: AuthDto): Promise<SearchExploreItem<AssetResponseDto>[]> {
+  async getExploreData(auth: AuthDto) {
     const options = { maxFields: 12, minAssetsPerField: 5 };
     const cities = await this.assetRepository.getAssetIdByCity(auth.user.id, options);
     const assets = await this.assetRepository.getByIdsWithAllRelationsButStacks(cities.items.map(({ data }) => data));

server/src/services/smart-info.service.spec.ts

@@ -1,11 +1,10 @@
 import { SystemConfig } from 'src/config';
 import { ImmichWorker, JobName, JobStatus } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { SmartInfoService } from 'src/services/smart-info.service';
 import { getCLIPModelInfo } from 'src/utils/misc';
 import { assetStub } from 'test/fixtures/asset.stub';
 import { systemConfigStub } from 'test/fixtures/system-config.stub';
-import { newTestService, ServiceMocks } from 'test/utils';
+import { makeStream, newTestService, ServiceMocks } from 'test/utils';
 
 describe(SmartInfoService.name, () => {
   let sut: SmartInfoService;
@@ -58,10 +57,6 @@ describe(SmartInfoService.name, () => {
       expect(mocks.search.getDimensionSize).not.toHaveBeenCalled();
       expect(mocks.search.setDimensionSize).not.toHaveBeenCalled();
       expect(mocks.search.deleteAllSearchEmbeddings).not.toHaveBeenCalled();
-      expect(mocks.job.getQueueStatus).not.toHaveBeenCalled();
-      expect(mocks.job.pause).not.toHaveBeenCalled();
-      expect(mocks.job.waitForQueueCompletion).not.toHaveBeenCalled();
-      expect(mocks.job.resume).not.toHaveBeenCalled();
     });
 
     it('should return if model and DB dimension size are equal', async () => {
@@ -72,38 +67,15 @@ describe(SmartInfoService.name, () => {
       expect(mocks.search.getDimensionSize).toHaveBeenCalledTimes(1);
       expect(mocks.search.setDimensionSize).not.toHaveBeenCalled();
       expect(mocks.search.deleteAllSearchEmbeddings).not.toHaveBeenCalled();
-      expect(mocks.job.getQueueStatus).not.toHaveBeenCalled();
-      expect(mocks.job.pause).not.toHaveBeenCalled();
-      expect(mocks.job.waitForQueueCompletion).not.toHaveBeenCalled();
-      expect(mocks.job.resume).not.toHaveBeenCalled();
     });
 
     it('should update DB dimension size if model and DB have different values', async () => {
       mocks.search.getDimensionSize.mockResolvedValue(768);
-      mocks.job.getQueueStatus.mockResolvedValue({ isActive: false, isPaused: false });
 
       await sut.onConfigInit({ newConfig: systemConfigStub.machineLearningEnabled as SystemConfig });
 
       expect(mocks.search.getDimensionSize).toHaveBeenCalledTimes(1);
       expect(mocks.search.setDimensionSize).toHaveBeenCalledWith(512);
-      expect(mocks.job.getQueueStatus).toHaveBeenCalledTimes(1);
-      expect(mocks.job.pause).toHaveBeenCalledTimes(1);
-      expect(mocks.job.waitForQueueCompletion).toHaveBeenCalledTimes(1);
-      expect(mocks.job.resume).toHaveBeenCalledTimes(1);
-    });
-
-    it('should skip pausing and resuming queue if already paused', async () => {
-      mocks.search.getDimensionSize.mockResolvedValue(768);
-      mocks.job.getQueueStatus.mockResolvedValue({ isActive: false, isPaused: true });
-
-      await sut.onConfigInit({ newConfig: systemConfigStub.machineLearningEnabled as SystemConfig });
-
-      expect(mocks.search.getDimensionSize).toHaveBeenCalledTimes(1);
-      expect(mocks.search.setDimensionSize).toHaveBeenCalledWith(512);
-      expect(mocks.job.getQueueStatus).toHaveBeenCalledTimes(1);
-      expect(mocks.job.pause).not.toHaveBeenCalled();
-      expect(mocks.job.waitForQueueCompletion).toHaveBeenCalledTimes(1);
-      expect(mocks.job.resume).not.toHaveBeenCalled();
     });
   });
 
@@ -120,10 +92,6 @@ describe(SmartInfoService.name, () => {
       expect(mocks.search.getDimensionSize).not.toHaveBeenCalled();
       expect(mocks.search.setDimensionSize).not.toHaveBeenCalled();
       expect(mocks.search.deleteAllSearchEmbeddings).not.toHaveBeenCalled();
-      expect(mocks.job.getQueueStatus).not.toHaveBeenCalled();
-      expect(mocks.job.pause).not.toHaveBeenCalled();
-      expect(mocks.job.waitForQueueCompletion).not.toHaveBeenCalled();
-      expect(mocks.job.resume).not.toHaveBeenCalled();
     });
 
     it('should return if model and DB dimension size are equal', async () => {
@@ -141,15 +109,10 @@ describe(SmartInfoService.name, () => {
       expect(mocks.search.getDimensionSize).toHaveBeenCalledTimes(1);
       expect(mocks.search.setDimensionSize).not.toHaveBeenCalled();
       expect(mocks.search.deleteAllSearchEmbeddings).not.toHaveBeenCalled();
-      expect(mocks.job.getQueueStatus).not.toHaveBeenCalled();
-      expect(mocks.job.pause).not.toHaveBeenCalled();
-      expect(mocks.job.waitForQueueCompletion).not.toHaveBeenCalled();
-      expect(mocks.job.resume).not.toHaveBeenCalled();
     });
 
     it('should update DB dimension size if model and DB have different values', async () => {
       mocks.search.getDimensionSize.mockResolvedValue(512);
-      mocks.job.getQueueStatus.mockResolvedValue({ isActive: false, isPaused: false });
 
       await sut.onConfigUpdate({
         newConfig: {
@@ -162,15 +125,10 @@ describe(SmartInfoService.name, () => {
 
       expect(mocks.search.getDimensionSize).toHaveBeenCalledTimes(1);
       expect(mocks.search.setDimensionSize).toHaveBeenCalledWith(768);
-      expect(mocks.job.getQueueStatus).toHaveBeenCalledTimes(1);
-      expect(mocks.job.pause).toHaveBeenCalledTimes(1);
-      expect(mocks.job.waitForQueueCompletion).toHaveBeenCalledTimes(1);
-      expect(mocks.job.resume).toHaveBeenCalledTimes(1);
     });
 
     it('should clear embeddings if old and new models are different', async () => {
       mocks.search.getDimensionSize.mockResolvedValue(512);
-      mocks.job.getQueueStatus.mockResolvedValue({ isActive: false, isPaused: false });
 
       await sut.onConfigUpdate({
         newConfig: {
@@ -184,31 +142,6 @@ describe(SmartInfoService.name, () => {
       expect(mocks.search.deleteAllSearchEmbeddings).toHaveBeenCalled();
       expect(mocks.search.getDimensionSize).toHaveBeenCalledTimes(1);
       expect(mocks.search.setDimensionSize).not.toHaveBeenCalled();
-      expect(mocks.job.getQueueStatus).toHaveBeenCalledTimes(1);
-      expect(mocks.job.pause).toHaveBeenCalledTimes(1);
-      expect(mocks.job.waitForQueueCompletion).toHaveBeenCalledTimes(1);
-      expect(mocks.job.resume).toHaveBeenCalledTimes(1);
-    });
-
-    it('should skip pausing and resuming queue if already paused', async () => {
-      mocks.search.getDimensionSize.mockResolvedValue(512);
-      mocks.job.getQueueStatus.mockResolvedValue({ isActive: false, isPaused: true });
-
-      await sut.onConfigUpdate({
-        newConfig: {
-          machineLearning: { clip: { modelName: 'ViT-B-32__openai', enabled: true }, enabled: true },
-        } as SystemConfig,
-        oldConfig: {
-          machineLearning: { clip: { modelName: 'ViT-B-16__openai', enabled: true }, enabled: true },
-        } as SystemConfig,
-      });
-
-      expect(mocks.search.getDimensionSize).toHaveBeenCalledTimes(1);
-      expect(mocks.search.setDimensionSize).not.toHaveBeenCalled();
-      expect(mocks.job.getQueueStatus).toHaveBeenCalledTimes(1);
-      expect(mocks.job.pause).not.toHaveBeenCalled();
-      expect(mocks.job.waitForQueueCompletion).toHaveBeenCalledTimes(1);
-      expect(mocks.job.resume).not.toHaveBeenCalled();
     });
   });
 
@@ -218,38 +151,31 @@ describe(SmartInfoService.name, () => {
 
       await sut.handleQueueEncodeClip({});
 
-      expect(mocks.asset.getAll).not.toHaveBeenCalled();
-      expect(mocks.asset.getWithout).not.toHaveBeenCalled();
+      expect(mocks.search.setDimensionSize).not.toHaveBeenCalled();
     });
 
     it('should queue the assets without clip embeddings', async () => {
-      mocks.asset.getWithout.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForEncodeClip.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueEncodeClip({ force: false });
 
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         { name: JobName.SMART_SEARCH, data: { id: assetStub.image.id } },
       ]);
-      expect(mocks.asset.getWithout).toHaveBeenCalledWith({ skip: 0, take: 1000 }, WithoutProperty.SMART_SEARCH);
-      expect(mocks.search.deleteAllSearchEmbeddings).not.toHaveBeenCalled();
+      expect(mocks.assetJob.streamForEncodeClip).toHaveBeenCalledWith(false);
+      expect(mocks.search.setDimensionSize).not.toHaveBeenCalled();
     });
 
     it('should queue all the assets', async () => {
-      mocks.asset.getAll.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForEncodeClip.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueEncodeClip({ force: true });
 
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         { name: JobName.SMART_SEARCH, data: { id: assetStub.image.id } },
       ]);
-      expect(mocks.asset.getAll).toHaveBeenCalled();
-      expect(mocks.search.deleteAllSearchEmbeddings).toHaveBeenCalled();
+      expect(mocks.assetJob.streamForEncodeClip).toHaveBeenCalledWith(true);
+      expect(mocks.search.setDimensionSize).toHaveBeenCalledExactlyOnceWith(512);
     });
   });
 

server/src/services/smart-info.service.ts

@@ -3,12 +3,10 @@ import { SystemConfig } from 'src/config';
 import { JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
 import { OnEvent, OnJob } from 'src/decorators';
 import { DatabaseLock, ImmichWorker, JobName, JobStatus, QueueName } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { ArgOf } from 'src/repositories/event.repository';
 import { BaseService } from 'src/services/base.service';
-import { JobOf } from 'src/types';
+import { JobItem, JobOf } from 'src/types';
 import { getCLIPModelInfo, isSmartSearchEnabled } from 'src/utils/misc';
-import { usePagination } from 'src/utils/pagination';
 
 @Injectable()
 export class SmartInfoService extends BaseService {
@@ -50,12 +48,6 @@ export class SmartInfoService extends BaseService {
         return;
       }
 
-      const { isPaused } = await this.jobRepository.getQueueStatus(QueueName.SMART_SEARCH);
-      if (!isPaused) {
-        await this.jobRepository.pause(QueueName.SMART_SEARCH);
-      }
-      await this.jobRepository.waitForQueueCompletion(QueueName.SMART_SEARCH);
-
       if (dimSizeChange) {
         this.logger.log(
           `Dimension size of model ${newConfig.machineLearning.clip.modelName} is ${dimSize}, but database expects ${dbDimSize}.`,
@@ -67,9 +59,8 @@ export class SmartInfoService extends BaseService {
         await this.searchRepository.deleteAllSearchEmbeddings();
       }
 
-      if (!isPaused) {
-        await this.jobRepository.resume(QueueName.SMART_SEARCH);
-      }
+      // TODO: A job to reindex all assets should be scheduled, though user
+      // confirmation should probably be requested before doing that.
     });
   }
 
@@ -81,21 +72,23 @@ export class SmartInfoService extends BaseService {
     }
 
     if (force) {
-      await this.searchRepository.deleteAllSearchEmbeddings();
+      const { dimSize } = getCLIPModelInfo(machineLearning.clip.modelName);
+      // in addition to deleting embeddings, update the dimension size in case it failed earlier
+      await this.searchRepository.setDimensionSize(dimSize);
     }
 
-    const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) => {
-      return force
-        ? this.assetRepository.getAll(pagination, { isVisible: true })
-        : this.assetRepository.getWithout(pagination, WithoutProperty.SMART_SEARCH);
-    });
-
-    for await (const assets of assetPagination) {
-      await this.jobRepository.queueAll(
-        assets.map((asset) => ({ name: JobName.SMART_SEARCH, data: { id: asset.id } })),
-      );
+    let queue: JobItem[] = [];
+    const assets = this.assetJobRepository.streamForEncodeClip(force);
+    for await (const asset of assets) {
+      queue.push({ name: JobName.SMART_SEARCH, data: { id: asset.id } });
+      if (queue.length >= JOBS_ASSET_PAGINATION_SIZE) {
+        await this.jobRepository.queueAll(queue);
+        queue = [];
+      }
     }
 
+    await this.jobRepository.queueAll(queue);
+
     return JobStatus.SUCCESS;
   }
 
@@ -126,6 +119,12 @@ export class SmartInfoService extends BaseService {
       await this.databaseRepository.wait(DatabaseLock.CLIPDimSize);
     }
 
+    const newConfig = await this.getConfig({ withCache: true });
+    if (machineLearning.clip.modelName !== newConfig.machineLearning.clip.modelName) {
+      // Skip the job if the the model has changed since the embedding was generated.
+      return JobStatus.SKIPPED;
+    }
+
     await this.searchRepository.upsert(asset.id, embedding);
 
     return JobStatus.SUCCESS;

server/src/services/storage-template.service.ts

@@ -116,6 +116,11 @@ export class StorageTemplateService extends BaseService {
     return { ...storageTokens, presetOptions: storagePresets };
   }
 
+  @OnEvent({ name: 'asset.metadataExtracted' })
+  async onAssetMetadataExtracted({ source, assetId }: ArgOf<'asset.metadataExtracted'>) {
+    await this.jobRepository.queue({ name: JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE, data: { source, id: assetId } });
+  }
+
   @OnJob({ name: JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE, queue: QueueName.STORAGE_TEMPLATE_MIGRATION })
   async handleMigrationSingle({ id }: JobOf<JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE>): Promise<JobStatus> {
     const config = await this.getConfig({ withCache: true });

server/src/services/system-config.service.spec.ts

@@ -6,6 +6,7 @@ import {
   CQMode,
   ImageFormat,
   LogLevel,
+  OAuthTokenEndpointAuthMethod,
   QueueName,
   ToneMapping,
   TranscodeHWAccel,
@@ -119,6 +120,8 @@ const updatedConfig = Object.freeze<SystemConfig>({
     scope: 'openid email profile',
     signingAlgorithm: 'RS256',
     profileSigningAlgorithm: 'none',
+    tokenEndpointAuthMethod: OAuthTokenEndpointAuthMethod.CLIENT_SECRET_POST,
+    timeout: 30_000,
     storageLabelClaim: 'preferred_username',
     storageQuotaClaim: 'immich_quota',
   },

server/src/types.ts

@@ -177,9 +177,10 @@ export interface IDelayedJob extends IBaseJob {
   delay?: number;
 }
 
+export type JobSource = 'upload' | 'sidecar-write' | 'copy';
 export interface IEntityJob extends IBaseJob {
   id: string;
-  source?: 'upload' | 'sidecar-write' | 'copy';
+  source?: JobSource;
   notify?: boolean;
 }
 
@@ -251,7 +252,7 @@ export interface INotifyAlbumInviteJob extends IEntityJob {
 }
 
 export interface INotifyAlbumUpdateJob extends IEntityJob, IDelayedJob {
-  recipientIds: string[];
+  recipientId: string;
 }
 
 export interface JobCounts {

server/src/utils/database.ts

@@ -18,10 +18,10 @@ import { parse } from 'pg-connection-string';
 import postgres, { Notice } from 'postgres';
 import { columns, Exif, Person } from 'src/database';
 import { DB } from 'src/db';
-import { AssetFileType } from 'src/enum';
+import { AssetFileType, DatabaseExtension } from 'src/enum';
 import { TimeBucketSize } from 'src/repositories/asset.repository';
 import { AssetSearchBuilderOptions } from 'src/repositories/search.repository';
-import { DatabaseConnectionParams } from 'src/types';
+import { DatabaseConnectionParams, VectorExtension } from 'src/types';
 
 type Ssl = 'require' | 'allow' | 'prefer' | 'verify-full' | boolean | object;
 
@@ -379,3 +379,28 @@ export function searchAssetBuilder(kysely: Kysely<DB>, options: AssetSearchBuild
     .$if(!!(options.withFaces || options.withPeople || options.personIds), (qb) => qb.select(withFacesAndPeople))
     .$if(!options.withDeleted, (qb) => qb.where('assets.deletedAt', 'is', null));
 }
+
+type VectorIndexOptions = { vectorExtension: VectorExtension; table: string; indexName: string };
+
+export function vectorIndexQuery({ vectorExtension, table, indexName }: VectorIndexOptions): string {
+  switch (vectorExtension) {
+    case DatabaseExtension.VECTORS: {
+      return `
+        CREATE INDEX IF NOT EXISTS ${indexName} ON ${table}
+        USING vectors (embedding vector_cos_ops) WITH (options = $$
+        [indexing.hnsw]
+        m = 16
+        ef_construction = 300
+        $$)`;
+    }
+    case DatabaseExtension.VECTOR: {
+      return `
+        CREATE INDEX IF NOT EXISTS ${indexName} ON ${table}
+        USING hnsw (embedding vector_cosine_ops)
+        WITH (ef_construction = 300, m = 16)`;
+    }
+    default: {
+      throw new Error(`Unsupported vector extension: '${vectorExtension}'`);
+    }
+  }
+}

server/src/utils/pagination.ts

@@ -8,22 +8,6 @@ export interface PaginationResult<T> {
   hasNextPage: boolean;
 }
 
-export type Paginated<T> = Promise<PaginationResult<T>>;
-
-/** @deprecated use `this.db. ... .stream()` instead */
-export async function* usePagination<T>(
-  pageSize: number,
-  getNextPage: (pagination: PaginationOptions) => PaginationResult<T> | Paginated<T>,
-) {
-  let hasNextPage = true;
-
-  for (let skip = 0; hasNextPage; skip += pageSize) {
-    const result = await getNextPage({ take: pageSize, skip });
-    hasNextPage = result.hasNextPage;
-    yield result.items;
-  }
-}
-
 export function paginationHelper<Entity extends object>(items: Entity[], take: number): PaginationResult<Entity> {
   const hasNextPage = items.length > take;
   items.splice(take);