refactor(server): Move metadata extraction to domain (#4243)

* use storageRepository in metadata extraction

* move metadata extraction processor to domain

* cleanup infra/domain

---------

Co-authored-by: Jason Rasmussen <jrasm91@gmail.com>

server/package.json

@@ -148,7 +148,7 @@
     "coverageDirectory": "./coverage",
     "coverageThreshold": {
       "./src/domain/": {
-        "branches": 80,
+        "branches": 75,
         "functions": 80,
         "lines": 90,
         "statements": 90

server/src/domain/metadata/geocoding.repository.ts

@@ -1,20 +0,0 @@
-import { InitOptions } from 'local-reverse-geocoder';
-
-export const IGeocodingRepository = 'IGeocodingRepository';
-
-export interface GeoPoint {
-  latitude: number;
-  longitude: number;
-}
-
-export interface ReverseGeocodeResult {
-  country: string | null;
-  state: string | null;
-  city: string | null;
-}
-
-export interface IGeocodingRepository {
-  init(options: Partial<InitOptions>): Promise<void>;
-  reverseGeocode(point: GeoPoint): Promise<ReverseGeocodeResult>;
-  deleteCache(): Promise<void>;
-}

server/src/domain/metadata/index.ts

@@ -1,2 +1,2 @@
-export * from './geocoding.repository';
+export * from './metadata.repository';
 export * from './metadata.service';

server/src/domain/metadata/metadata.repository.ts

@@ -0,0 +1,31 @@
+import { Tags } from 'exiftool-vendored';
+import { InitOptions } from 'local-reverse-geocoder';
+
+export const IMetadataRepository = 'IMetadataRepository';
+
+export interface GeoPoint {
+  latitude: number;
+  longitude: number;
+}
+
+export interface ReverseGeocodeResult {
+  country: string | null;
+  state: string | null;
+  city: string | null;
+}
+
+export interface ImmichTags extends Tags {
+  ContentIdentifier?: string;
+  MotionPhoto?: number;
+  MotionPhotoVersion?: number;
+  MotionPhotoPresentationTimestampUs?: number;
+  MediaGroupUUID?: string;
+  ImagePixelDepth?: string;
+}
+
+export interface IMetadataRepository {
+  init(options: Partial<InitOptions>): Promise<void>;
+  reverseGeocode(point: GeoPoint): Promise<ReverseGeocodeResult>;
+  deleteCache(): Promise<void>;
+  getExifTags(path: string): Promise<ImmichTags | null>;
+}

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

@@ -1,22 +1,43 @@
-import { assetStub, newAssetRepositoryMock, newJobRepositoryMock, newStorageRepositoryMock } from '@test';
+import {
+  assetStub,
+  newAlbumRepositoryMock,
+  newAssetRepositoryMock,
+  newCryptoRepositoryMock,
+  newJobRepositoryMock,
+  newMetadataRepositoryMock,
+  newStorageRepositoryMock,
+  newSystemConfigRepositoryMock,
+} from '@test';
 import { constants } from 'fs/promises';
+import { IAlbumRepository } from '../album';
 import { IAssetRepository, WithProperty, WithoutProperty } from '../asset';
+import { ICryptoRepository } from '../crypto';
 import { IJobRepository, JobName } from '../job';
 import { IStorageRepository } from '../storage';
+import { ISystemConfigRepository } from '../system-config';
+import { IMetadataRepository } from './metadata.repository';
 import { MetadataService } from './metadata.service';
 
 describe(MetadataService.name, () => {
-  let sut: MetadataService;
+  let albumMock: jest.Mocked<IAlbumRepository>;
   let assetMock: jest.Mocked<IAssetRepository>;
+  let configMock: jest.Mocked<ISystemConfigRepository>;
+  let cryptoRepository: jest.Mocked<ICryptoRepository>;
   let jobMock: jest.Mocked<IJobRepository>;
+  let metadataMock: jest.Mocked<IMetadataRepository>;
   let storageMock: jest.Mocked<IStorageRepository>;
+  let sut: MetadataService;
 
   beforeEach(async () => {
+    albumMock = newAlbumRepositoryMock();
     assetMock = newAssetRepositoryMock();
+    configMock = newSystemConfigRepositoryMock();
+    cryptoRepository = newCryptoRepositoryMock();
     jobMock = newJobRepositoryMock();
+    metadataMock = newMetadataRepositoryMock();
     storageMock = newStorageRepositoryMock();
 
-    sut = new MetadataService(assetMock, jobMock, storageMock);
+    sut = new MetadataService(albumMock, assetMock, cryptoRepository, jobMock, metadataMock, storageMock, configMock);
   });
 
   it('should be defined', () => {

server/src/domain/metadata/metadata.service.ts

@@ -1,16 +1,148 @@
-import { Inject } from '@nestjs/common';
+import { AssetEntity, AssetType, ExifEntity } from '@app/infra/entities';
+import { Inject, Injectable, Logger } from '@nestjs/common';
+import { ExifDateTime } from 'exiftool-vendored';
+import { firstDateTime } from 'exiftool-vendored/dist/FirstDateTime';
 import { constants } from 'fs/promises';
-import { IAssetRepository, WithoutProperty, WithProperty } from '../asset';
+import { Duration } from 'luxon';
+import { IAlbumRepository } from '../album';
+import { IAssetRepository, WithProperty, WithoutProperty } from '../asset';
+import { ICryptoRepository } from '../crypto';
 import { usePagination } from '../domain.util';
-import { IBaseJob, IEntityJob, IJobRepository, JobName, JOBS_ASSET_PAGINATION_SIZE } from '../job';
+import { IBaseJob, IEntityJob, IJobRepository, JOBS_ASSET_PAGINATION_SIZE, JobName, QueueName } from '../job';
-import { IStorageRepository } from '../storage';
+import { IStorageRepository, StorageCore, StorageFolder } from '../storage';
+import { FeatureFlag, ISystemConfigRepository, SystemConfigCore } from '../system-config';
+import { IMetadataRepository, ImmichTags } from './metadata.repository';
 
+interface DirectoryItem {
+  Length?: number;
+  Mime: string;
+  Padding?: number;
+  Semantic?: string;
+}
+
+interface DirectoryEntry {
+  Item: DirectoryItem;
+}
+
+const exifDate = (dt: ExifDateTime | string | undefined) => (dt instanceof ExifDateTime ? dt?.toDate() : null);
+// exiftool returns strings when it fails to parse non-string values, so this is used where a string is not expected
+const validate = <T>(value: T): T | null => (typeof value === 'string' ? null : value ?? null);
+
+@Injectable()
 export class MetadataService {
+  private logger = new Logger(MetadataService.name);
+  private storageCore: StorageCore;
+  private configCore: SystemConfigCore;
+  private oldCities?: string;
+
   constructor(
+    @Inject(IAlbumRepository) private albumRepository: IAlbumRepository,
     @Inject(IAssetRepository) private assetRepository: IAssetRepository,
+    @Inject(ICryptoRepository) private cryptoRepository: ICryptoRepository,
     @Inject(IJobRepository) private jobRepository: IJobRepository,
+    @Inject(IMetadataRepository) private repository: IMetadataRepository,
     @Inject(IStorageRepository) private storageRepository: IStorageRepository,
-  ) {}
+    @Inject(ISystemConfigRepository) configRepository: ISystemConfigRepository,
+  ) {
+    this.storageCore = new StorageCore(storageRepository);
+    this.configCore = new SystemConfigCore(configRepository);
+    this.configCore.config$.subscribe(() => this.init());
+  }
+
+  async init(deleteCache = false) {
+    const { reverseGeocoding } = await this.configCore.getConfig();
+    const { citiesFileOverride } = reverseGeocoding;
+
+    if (!reverseGeocoding.enabled) {
+      return;
+    }
+
+    try {
+      if (deleteCache) {
+        await this.repository.deleteCache();
+      } else if (this.oldCities && this.oldCities === citiesFileOverride) {
+        return;
+      }
+
+      await this.jobRepository.pause(QueueName.METADATA_EXTRACTION);
+      await this.repository.init({ citiesFileOverride });
+      await this.jobRepository.resume(QueueName.METADATA_EXTRACTION);
+
+      this.logger.log(`Initialized local reverse geocoder with ${citiesFileOverride}`);
+      this.oldCities = citiesFileOverride;
+    } catch (error: Error | any) {
+      this.logger.error(`Unable to initialize reverse geocoding: ${error}`, error?.stack);
+    }
+  }
+
+  async handleLivePhotoLinking(job: IEntityJob) {
+    const { id } = job;
+    const [asset] = await this.assetRepository.getByIds([id]);
+    if (!asset?.exifInfo) {
+      return false;
+    }
+
+    if (!asset.exifInfo.livePhotoCID) {
+      return true;
+    }
+
+    const otherType = asset.type === AssetType.VIDEO ? AssetType.IMAGE : AssetType.VIDEO;
+    const match = await this.assetRepository.findLivePhotoMatch({
+      livePhotoCID: asset.exifInfo.livePhotoCID,
+      ownerId: asset.ownerId,
+      otherAssetId: asset.id,
+      type: otherType,
+    });
+
+    if (!match) {
+      return true;
+    }
+
+    const [photoAsset, motionAsset] = asset.type === AssetType.IMAGE ? [asset, match] : [match, asset];
+
+    await this.assetRepository.save({ id: photoAsset.id, livePhotoVideoId: motionAsset.id });
+    await this.assetRepository.save({ id: motionAsset.id, isVisible: false });
+    await this.albumRepository.removeAsset(motionAsset.id);
+
+    return true;
+  }
+
+  async handleQueueMetadataExtraction(job: IBaseJob) {
+    const { force } = job;
+    const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) => {
+      return force
+        ? this.assetRepository.getAll(pagination)
+        : this.assetRepository.getWithout(pagination, WithoutProperty.EXIF);
+    });
+
+    for await (const assets of assetPagination) {
+      for (const asset of assets) {
+        await this.jobRepository.queue({ name: JobName.METADATA_EXTRACTION, data: { id: asset.id } });
+      }
+    }
+
+    return true;
+  }
+
+  async handleMetadataExtraction({ id }: IEntityJob) {
+    const [asset] = await this.assetRepository.getByIds([id]);
+    if (!asset || !asset.isVisible) {
+      return false;
+    }
+
+    const { exifData, tags } = await this.exifData(asset);
+
+    await this.applyMotionPhotos(asset, tags);
+    await this.applyReverseGeocoding(asset, exifData);
+    await this.assetRepository.upsertExif(exifData);
+    await this.assetRepository.save({
+      id: asset.id,
+      duration: tags.Duration ? this.getDuration(tags.Duration) : null,
+      fileCreatedAt: exifData.dateTimeOriginal ?? undefined,
+    });
+
+    return true;
+  }
 
   async handleQueueSidecar(job: IBaseJob) {
     const { force } = job;
@@ -51,4 +183,156 @@ export class MetadataService {
 
     return true;
   }
+
+  private async applyReverseGeocoding(asset: AssetEntity, exifData: ExifEntity) {
+    const { latitude, longitude } = exifData;
+    if (!(await this.configCore.hasFeature(FeatureFlag.REVERSE_GEOCODING)) || !longitude || !latitude) {
+      return;
+    }
+
+    try {
+      const { city, state, country } = await this.repository.reverseGeocode({ latitude, longitude });
+      Object.assign(exifData, { city, state, country });
+    } catch (error: Error | any) {
+      this.logger.warn(
+        `Unable to run reverse geocoding due to ${error} for asset ${asset.id} at ${asset.originalPath}`,
+        error?.stack,
+      );
+    }
+  }
+
+  private async applyMotionPhotos(asset: AssetEntity, tags: ImmichTags) {
+    if (asset.type !== AssetType.IMAGE || asset.livePhotoVideoId) {
+      return;
+    }
+
+    const rawDirectory = tags.Directory;
+    const isMotionPhoto = tags.MotionPhoto;
+    const isMicroVideo = tags.MicroVideo;
+    const videoOffset = tags.MicroVideoOffset;
+    const directory = Array.isArray(rawDirectory) ? (rawDirectory as DirectoryEntry[]) : null;
+
+    let length = 0;
+    let padding = 0;
+
+    if (isMotionPhoto && directory) {
+      for (const entry of directory) {
+        if (entry.Item.Semantic == 'MotionPhoto') {
+          length = entry.Item.Length ?? 0;
+          padding = entry.Item.Padding ?? 0;
+          break;
+        }
+      }
+    }
+
+    if (isMicroVideo && typeof videoOffset === 'number') {
+      length = videoOffset;
+    }
+
+    if (!length) {
+      return;
+    }
+
+    this.logger.debug(`Starting motion photo video extraction (${asset.id})`);
+
+    try {
+      const stat = await this.storageRepository.stat(asset.originalPath);
+      const position = stat.size - length - padding;
+      const video = await this.storageRepository.readFile(asset.originalPath, {
+        buffer: Buffer.alloc(length),
+        position,
+        length,
+      });
+      const checksum = await this.cryptoRepository.hashSha1(video);
+
+      let motionAsset = await this.assetRepository.getByChecksum(asset.ownerId, checksum);
+      if (!motionAsset) {
+        motionAsset = await this.assetRepository.save({
+          libraryId: asset.libraryId,
+          type: AssetType.VIDEO,
+          fileCreatedAt: asset.fileCreatedAt ?? asset.createdAt,
+          fileModifiedAt: asset.fileModifiedAt,
+          checksum,
+          ownerId: asset.ownerId,
+          originalPath: this.storageCore.ensurePath(StorageFolder.ENCODED_VIDEO, asset.ownerId, `${asset.id}-MP.mp4`),
+          originalFileName: asset.originalFileName,
+          isVisible: false,
+          isReadOnly: true,
+          deviceAssetId: 'NONE',
+          deviceId: 'NONE',
+        });
+
+        await this.storageRepository.writeFile(asset.originalPath, video);
+
+        await this.jobRepository.queue({ name: JobName.METADATA_EXTRACTION, data: { id: motionAsset.id } });
+      }
+
+      await this.assetRepository.save({ id: asset.id, livePhotoVideoId: motionAsset.id });
+
+      this.logger.debug(`Finished motion photo video extraction (${asset.id})`);
+    } catch (error: Error | any) {
+      this.logger.error(`Failed to extract live photo ${asset.originalPath}: ${error}`, error?.stack);
+    }
+  }
+
+  private async exifData(asset: AssetEntity): Promise<{ exifData: ExifEntity; tags: ImmichTags }> {
+    const stats = await this.storageRepository.stat(asset.originalPath);
+    const mediaTags = await this.repository.getExifTags(asset.originalPath);
+    const sidecarTags = asset.sidecarPath ? await this.repository.getExifTags(asset.sidecarPath) : null;
+    const tags = { ...mediaTags, ...sidecarTags };
+
+    this.logger.verbose('Exif Tags', tags);
+
+    return {
+      exifData: <ExifEntity>{
+        // altitude: tags.GPSAltitude ?? null,
+        assetId: asset.id,
+        bitsPerSample: this.getBitsPerSample(tags),
+        colorspace: tags.ColorSpace ?? null,
+        dateTimeOriginal: exifDate(firstDateTime(tags)) ?? asset.fileCreatedAt,
+        exifImageHeight: validate(tags.ImageHeight),
+        exifImageWidth: validate(tags.ImageWidth),
+        exposureTime: tags.ExposureTime ?? null,
+        fileSizeInByte: stats.size,
+        fNumber: validate(tags.FNumber),
+        focalLength: validate(tags.FocalLength),
+        fps: validate(tags.VideoFrameRate),
+        iso: validate(tags.ISO),
+        latitude: validate(tags.GPSLatitude),
+        lensModel: tags.LensModel ?? null,
+        livePhotoCID: (asset.type === AssetType.VIDEO ? tags.ContentIdentifier : tags.MediaGroupUUID) ?? null,
+        longitude: validate(tags.GPSLongitude),
+        make: tags.Make ?? null,
+        model: tags.Model ?? null,
+        modifyDate: exifDate(tags.ModifyDate) ?? asset.fileModifiedAt,
+        orientation: validate(tags.Orientation)?.toString() ?? null,
+        profileDescription: tags.ProfileDescription || tags.ProfileName || null,
+        projectionType: tags.ProjectionType ? String(tags.ProjectionType).toUpperCase() : null,
+        timeZone: tags.tz,
+      },
+      tags,
+    };
+  }
+
+  private getBitsPerSample(tags: ImmichTags): number | null {
+    const bitDepthTags = [
+      tags.BitsPerSample,
+      tags.ComponentBitDepth,
+      tags.ImagePixelDepth,
+      tags.BitDepth,
+      tags.ColorBitDepth,
+      // `numericTags` doesn't parse values like '12 12 12'
+    ].map((tag) => (typeof tag === 'string' ? Number.parseInt(tag) : tag));
+
+    let bitsPerSample = bitDepthTags.find((tag) => typeof tag === 'number' && !Number.isNaN(tag)) ?? null;
+    if (bitsPerSample && bitsPerSample >= 24 && bitsPerSample % 3 === 0) {
+      bitsPerSample /= 3; // converts per-pixel bit depth to per-channel
+    }
+
+    return bitsPerSample;
+  }
+
+  private getDuration(seconds?: number): string {
+    return Duration.fromObject({ seconds }).toFormat('hh:mm:ss.SSS');
+  }
 }

server/src/domain/storage/storage.repository.ts

@@ -1,4 +1,5 @@
 import { Stats } from 'fs';
+import { FileReadOptions } from 'fs/promises';
 import { Readable } from 'stream';
 import { CrawlOptionsDto } from '../library';
 
@@ -24,6 +25,8 @@ export const IStorageRepository = 'IStorageRepository';
 export interface IStorageRepository {
   createZipStream(): ImmichZipStream;
   createReadStream(filepath: string, mimeType?: string | null): Promise<ImmichReadStream>;
+  readFile(filepath: string, options?: FileReadOptions<Buffer>): Promise<Buffer>;
+  writeFile(filepath: string, buffer: Buffer): Promise<void>;
   unlink(filepath: string): Promise<void>;
   unlinkDir(folder: string, options?: { recursive?: boolean; force?: boolean }): Promise<void>;
   removeEmptyDirs(folder: string, self?: boolean): Promise<void>;

server/src/infra/infra.module.ts

@@ -6,13 +6,12 @@ import {
   ICommunicationRepository,
   ICryptoRepository,
   IFaceRepository,
-  IGeocodingRepository,
   IJobRepository,
   IKeyRepository,
   ILibraryRepository,
   IMachineLearningRepository,
   IMediaRepository,
-  immichAppConfig,
+  IMetadataRepository,
   IPartnerRepository,
   IPersonRepository,
   ISearchRepository,
@@ -23,6 +22,7 @@ import {
   ITagRepository,
   IUserRepository,
   IUserTokenRepository,
+  immichAppConfig,
 } from '@app/domain';
 import { BullModule } from '@nestjs/bullmq';
 import { Global, Module, Provider } from '@nestjs/common';
@@ -33,20 +33,20 @@ import { databaseConfig } from './database.config';
 import { databaseEntities } from './entities';
 import { bullConfig, bullQueues } from './infra.config';
 import {
+  APIKeyRepository,
   AccessRepository,
   AlbumRepository,
-  APIKeyRepository,
   AssetRepository,
   AuditRepository,
   CommunicationRepository,
   CryptoRepository,
   FaceRepository,
   FilesystemProvider,
-  GeocodingRepository,
   JobRepository,
   LibraryRepository,
   MachineLearningRepository,
   MediaRepository,
+  MetadataRepository,
   PartnerRepository,
   PersonRepository,
   SharedLinkRepository,
@@ -66,11 +66,11 @@ const providers: Provider[] = [
   { provide: ICommunicationRepository, useClass: CommunicationRepository },
   { provide: ICryptoRepository, useClass: CryptoRepository },
   { provide: IFaceRepository, useClass: FaceRepository },
-  { provide: IGeocodingRepository, useClass: GeocodingRepository },
   { provide: IJobRepository, useClass: JobRepository },
   { provide: ILibraryRepository, useClass: LibraryRepository },
   { provide: IKeyRepository, useClass: APIKeyRepository },
   { provide: IMachineLearningRepository, useClass: MachineLearningRepository },
+  { provide: IMetadataRepository, useClass: MetadataRepository },
   { provide: IPartnerRepository, useClass: PartnerRepository },
   { provide: IPersonRepository, useClass: PersonRepository },
   { provide: ISearchRepository, useClass: TypesenseRepository },

server/src/infra/repositories/filesystem.provider.ts

@@ -8,7 +8,7 @@ import {
 } from '@app/domain';
 import archiver from 'archiver';
 import { constants, createReadStream, existsSync, mkdirSync } from 'fs';
-import fs, { readdir } from 'fs/promises';
+import fs, { readdir, writeFile } from 'fs/promises';
 import { glob } from 'glob';
 import mv from 'mv';
 import { promisify } from 'node:util';
@@ -39,6 +39,18 @@ export class FilesystemProvider implements IStorageRepository {
     };
   }
 
+  async readFile(filepath: string, options?: fs.FileReadOptions<Buffer>): Promise<Buffer> {
+    const file = await fs.open(filepath);
+    try {
+      const { buffer } = await file.read(options);
+      return buffer;
+    } finally {
+      await file.close();
+    }
+  }
+
+  writeFile = writeFile;
+
   async moveFile(source: string, destination: string): Promise<void> {
     if (await this.checkFileExists(destination)) {
       throw new Error(`Destination file already exists: ${destination}`);

server/src/infra/repositories/index.ts

@@ -7,11 +7,11 @@ export * from './communication.repository';
 export * from './crypto.repository';
 export * from './face.repository';
 export * from './filesystem.provider';
-export * from './geocoding.repository';
 export * from './job.repository';
 export * from './library.repository';
 export * from './machine-learning.repository';
 export * from './media.repository';
+export * from './metadata.repository';
 export * from './partner.repository';
 export * from './person.repository';
 export * from './shared-link.repository';

server/src/infra/repositories/metadata.repository.ts

@@ -1,7 +1,9 @@
-import { GeoPoint, IGeocodingRepository, ReverseGeocodeResult } from '@app/domain';
+import { GeoPoint, IMetadataRepository, ImmichTags, ReverseGeocodeResult } from '@app/domain';
 import { REVERSE_GEOCODING_DUMP_DIRECTORY } from '@app/infra';
 import { Injectable, Logger } from '@nestjs/common';
+import { DefaultReadTaskOptions, exiftool } from 'exiftool-vendored';
 import { readdir, rm } from 'fs/promises';
+import * as geotz from 'geo-tz';
 import { getName } from 'i18n-iso-countries';
 import geocoder, { AddressObject, InitOptions } from 'local-reverse-geocoder';
 import path from 'path';
@@ -21,8 +23,8 @@ export type GeoData = AddressObject & {
 const lookup = promisify<GeoPoint[], number, AddressObject[][]>(geocoder.lookUp).bind(geocoder);
 
 @Injectable()
-export class GeocodingRepository implements IGeocodingRepository {
+export class MetadataRepository implements IMetadataRepository {
-  private logger = new Logger(GeocodingRepository.name);
+  private logger = new Logger(MetadataRepository.name);
 
   async init(options: Partial<InitOptions>): Promise<void> {
     return new Promise<void>((resolve) => {
@@ -69,4 +71,22 @@ export class GeocodingRepository implements IGeocodingRepository {
 
     return { country, state, city };
   }
+
+  getExifTags(path: string): Promise<ImmichTags | null> {
+    return exiftool
+      .read<ImmichTags>(path, undefined, {
+        ...DefaultReadTaskOptions,
+
+        defaultVideosToUTC: true,
+        backfillTimezones: true,
+        inferTimezoneFromDatestamps: true,
+        useMWG: true,
+        numericTags: DefaultReadTaskOptions.numericTags.concat(['FocalLength']),
+        geoTz: (lat, lon) => geotz.find(lat, lon)[0],
+      })
+      .catch((error) => {
+        this.logger.warn(`Error reading exif data (${path}): ${error}`, error?.stack);
+        return null;
+      });
+  }
 }

server/src/microservices/app.service.ts

@@ -17,16 +17,12 @@ import {
 } from '@app/domain';
 
 import { Injectable, Logger } from '@nestjs/common';
-import { MetadataExtractionProcessor } from './processors/metadata-extraction.processor';
 
 @Injectable()
 export class AppService {
   private logger = new Logger(AppService.name);
 
   constructor(
-    // TODO refactor to domain
-    private metadataProcessor: MetadataExtractionProcessor,
-
     private facialRecognitionService: FacialRecognitionService,
     private jobService: JobService,
     private mediaService: MediaService,
@@ -73,9 +69,9 @@ export class AppService {
       [JobName.GENERATE_THUMBHASH_THUMBNAIL]: (data) => this.mediaService.handleGenerateThumbhashThumbnail(data),
       [JobName.QUEUE_VIDEO_CONVERSION]: (data) => this.mediaService.handleQueueVideoConversion(data),
       [JobName.VIDEO_CONVERSION]: (data) => this.mediaService.handleVideoConversion(data),
-      [JobName.QUEUE_METADATA_EXTRACTION]: (data) => this.metadataProcessor.handleQueueMetadataExtraction(data),
+      [JobName.QUEUE_METADATA_EXTRACTION]: (data) => this.metadataService.handleQueueMetadataExtraction(data),
-      [JobName.METADATA_EXTRACTION]: (data) => this.metadataProcessor.handleMetadataExtraction(data),
+      [JobName.METADATA_EXTRACTION]: (data) => this.metadataService.handleMetadataExtraction(data),
-      [JobName.LINK_LIVE_PHOTOS]: (data) => this.metadataProcessor.handleLivePhotoLinking(data),
+      [JobName.LINK_LIVE_PHOTOS]: (data) => this.metadataService.handleLivePhotoLinking(data),
       [JobName.QUEUE_RECOGNIZE_FACES]: (data) => this.facialRecognitionService.handleQueueRecognizeFaces(data),
       [JobName.RECOGNIZE_FACES]: (data) => this.facialRecognitionService.handleRecognizeFaces(data),
       [JobName.GENERATE_PERSON_THUMBNAIL]: (data) => this.facialRecognitionService.handleGeneratePersonThumbnail(data),
@@ -99,10 +95,10 @@ export class AppService {
       }
 
       this.logger.warn('Geocoding csv parse error, trying again without cache...');
-      this.metadataProcessor.init(true);
+      this.metadataService.init(true);
     });
 
-    await this.metadataProcessor.init();
+    await this.metadataService.init();
     await this.searchService.init();
   }
 }

server/src/microservices/microservices.module.ts

@@ -2,10 +2,9 @@ import { DomainModule } from '@app/domain';
 import { InfraModule } from '@app/infra';
 import { Module } from '@nestjs/common';
 import { AppService } from './app.service';
-import { MetadataExtractionProcessor } from './processors/metadata-extraction.processor';
 
 @Module({
   imports: [DomainModule.register({ imports: [InfraModule] })],
-  providers: [MetadataExtractionProcessor, AppService],
+  providers: [AppService],
 })
 export class MicroservicesModule {}

server/src/microservices/processors/metadata-extraction.processor.ts

@@ -1,345 +0,0 @@
-import {
-  FeatureFlag,
-  IAlbumRepository,
-  IAssetRepository,
-  IBaseJob,
-  ICryptoRepository,
-  IEntityJob,
-  IGeocodingRepository,
-  IJobRepository,
-  IStorageRepository,
-  ISystemConfigRepository,
-  JobName,
-  JOBS_ASSET_PAGINATION_SIZE,
-  QueueName,
-  StorageCore,
-  StorageFolder,
-  SystemConfigCore,
-  usePagination,
-  WithoutProperty,
-} from '@app/domain';
-import { AssetEntity, AssetType, ExifEntity } from '@app/infra/entities';
-import { Inject, Logger } from '@nestjs/common';
-import { DefaultReadTaskOptions, ExifDateTime, exiftool, ReadTaskOptions, Tags } from 'exiftool-vendored';
-import { firstDateTime } from 'exiftool-vendored/dist/FirstDateTime';
-import * as geotz from 'geo-tz';
-import { Duration } from 'luxon';
-import fs from 'node:fs/promises';
-import path from 'node:path';
-
-interface DirectoryItem {
-  Length?: number;
-  Mime: string;
-  Padding?: number;
-  Semantic?: string;
-}
-
-interface DirectoryEntry {
-  Item: DirectoryItem;
-}
-
-interface ImmichTags extends Tags {
-  ContentIdentifier?: string;
-  MotionPhoto?: number;
-  MotionPhotoVersion?: number;
-  MotionPhotoPresentationTimestampUs?: number;
-  MediaGroupUUID?: string;
-  ImagePixelDepth?: string;
-}
-
-const exifDate = (dt: ExifDateTime | string | undefined) => (dt instanceof ExifDateTime ? dt?.toDate() : null);
-// exiftool returns strings when it fails to parse non-string values, so this is used where a string is not expected
-const validate = <T>(value: T): T | null => (typeof value === 'string' ? null : value ?? null);
-
-export class MetadataExtractionProcessor {
-  private logger = new Logger(MetadataExtractionProcessor.name);
-  private storageCore: StorageCore;
-  private configCore: SystemConfigCore;
-  private oldCities?: string;
-
-  constructor(
-    @Inject(IAssetRepository) private assetRepository: IAssetRepository,
-    @Inject(IAlbumRepository) private albumRepository: IAlbumRepository,
-    @Inject(IJobRepository) private jobRepository: IJobRepository,
-    @Inject(IGeocodingRepository) private geocodingRepository: IGeocodingRepository,
-    @Inject(ICryptoRepository) private cryptoRepository: ICryptoRepository,
-    @Inject(IStorageRepository) private storageRepository: IStorageRepository,
-    @Inject(ISystemConfigRepository) configRepository: ISystemConfigRepository,
-  ) {
-    this.storageCore = new StorageCore(storageRepository);
-    this.configCore = new SystemConfigCore(configRepository);
-    this.configCore.config$.subscribe(() => this.init());
-  }
-
-  async init(deleteCache = false) {
-    const { reverseGeocoding } = await this.configCore.getConfig();
-    const { citiesFileOverride } = reverseGeocoding;
-
-    if (!reverseGeocoding.enabled) {
-      return;
-    }
-
-    try {
-      if (deleteCache) {
-        await this.geocodingRepository.deleteCache();
-      } else if (this.oldCities && this.oldCities === citiesFileOverride) {
-        return;
-      }
-
-      await this.jobRepository.pause(QueueName.METADATA_EXTRACTION);
-      await this.geocodingRepository.init({ citiesFileOverride });
-      await this.jobRepository.resume(QueueName.METADATA_EXTRACTION);
-
-      this.logger.log(`Initialized local reverse geocoder with ${citiesFileOverride}`);
-      this.oldCities = citiesFileOverride;
-    } catch (error: Error | any) {
-      this.logger.error(`Unable to initialize reverse geocoding: ${error}`, error?.stack);
-    }
-  }
-
-  async handleLivePhotoLinking(job: IEntityJob) {
-    const { id } = job;
-    const [asset] = await this.assetRepository.getByIds([id]);
-    if (!asset?.exifInfo) {
-      return false;
-    }
-
-    if (!asset.exifInfo.livePhotoCID) {
-      return true;
-    }
-
-    const otherType = asset.type === AssetType.VIDEO ? AssetType.IMAGE : AssetType.VIDEO;
-    const match = await this.assetRepository.findLivePhotoMatch({
-      livePhotoCID: asset.exifInfo.livePhotoCID,
-      ownerId: asset.ownerId,
-      otherAssetId: asset.id,
-      type: otherType,
-    });
-
-    if (!match) {
-      return true;
-    }
-
-    const [photoAsset, motionAsset] = asset.type === AssetType.IMAGE ? [asset, match] : [match, asset];
-
-    await this.assetRepository.save({ id: photoAsset.id, livePhotoVideoId: motionAsset.id });
-    await this.assetRepository.save({ id: motionAsset.id, isVisible: false });
-    await this.albumRepository.removeAsset(motionAsset.id);
-
-    return true;
-  }
-
-  async handleQueueMetadataExtraction(job: IBaseJob) {
-    const { force } = job;
-    const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) => {
-      return force
-        ? this.assetRepository.getAll(pagination)
-        : this.assetRepository.getWithout(pagination, WithoutProperty.EXIF);
-    });
-
-    for await (const assets of assetPagination) {
-      for (const asset of assets) {
-        await this.jobRepository.queue({ name: JobName.METADATA_EXTRACTION, data: { id: asset.id } });
-      }
-    }
-
-    return true;
-  }
-
-  async handleMetadataExtraction({ id }: IEntityJob) {
-    const [asset] = await this.assetRepository.getByIds([id]);
-    if (!asset || !asset.isVisible) {
-      return false;
-    }
-
-    const [exifData, tags] = await this.exifData(asset);
-
-    await this.applyMotionPhotos(asset, tags);
-    await this.applyReverseGeocoding(asset, exifData);
-    await this.assetRepository.upsertExif(exifData);
-    await this.assetRepository.save({
-      id: asset.id,
-      duration: tags.Duration ? Duration.fromObject({ seconds: tags.Duration }).toFormat('hh:mm:ss.SSS') : null,
-      fileCreatedAt: exifData.dateTimeOriginal ?? undefined,
-    });
-
-    return true;
-  }
-
-  private async applyReverseGeocoding(asset: AssetEntity, exifData: ExifEntity) {
-    const { latitude, longitude } = exifData;
-    if (!(await this.configCore.hasFeature(FeatureFlag.REVERSE_GEOCODING)) || !longitude || !latitude) {
-      return;
-    }
-
-    try {
-      const { city, state, country } = await this.geocodingRepository.reverseGeocode({ latitude, longitude });
-      Object.assign(exifData, { city, state, country });
-    } catch (error: Error | any) {
-      this.logger.warn(
-        `Unable to run reverse geocoding due to ${error} for asset ${asset.id} at ${asset.originalPath}`,
-        error?.stack,
-      );
-    }
-  }
-
-  private async applyMotionPhotos(asset: AssetEntity, tags: ImmichTags) {
-    if (asset.type !== AssetType.IMAGE || asset.livePhotoVideoId) {
-      return;
-    }
-
-    const rawDirectory = tags.Directory;
-    const isMotionPhoto = tags.MotionPhoto;
-    const isMicroVideo = tags.MicroVideo;
-    const videoOffset = tags.MicroVideoOffset;
-    const directory = Array.isArray(rawDirectory) ? (rawDirectory as DirectoryEntry[]) : null;
-
-    let length = 0;
-    let padding = 0;
-
-    if (isMotionPhoto && directory) {
-      for (const entry of directory) {
-        if (entry.Item.Semantic == 'MotionPhoto') {
-          length = entry.Item.Length ?? 0;
-          padding = entry.Item.Padding ?? 0;
-          break;
-        }
-      }
-    }
-
-    if (isMicroVideo && typeof videoOffset === 'number') {
-      length = videoOffset;
-    }
-
-    if (!length) {
-      return;
-    }
-
-    this.logger.debug(`Starting motion photo video extraction (${asset.id})`);
-
-    let file = null;
-    try {
-      const encodedFolder = this.storageCore.getFolderLocation(StorageFolder.ENCODED_VIDEO, asset.ownerId);
-      const encodedFile = path.join(encodedFolder, path.parse(asset.originalPath).name + '.mp4');
-      this.storageRepository.mkdirSync(encodedFolder);
-
-      file = await fs.open(asset.originalPath);
-
-      const stat = await file.stat();
-      const position = stat.size - length - padding;
-      const video = await file.read({ buffer: Buffer.alloc(length), position, length });
-      const checksum = await this.cryptoRepository.hashSha1(video.buffer);
-
-      let motionAsset = await this.assetRepository.getByChecksum(asset.ownerId, checksum);
-      if (!motionAsset) {
-        motionAsset = await this.assetRepository.save({
-          libraryId: asset.libraryId,
-          type: AssetType.VIDEO,
-          fileCreatedAt: asset.fileCreatedAt ?? asset.createdAt,
-          fileModifiedAt: asset.fileModifiedAt,
-          checksum,
-          ownerId: asset.ownerId,
-          originalPath: encodedFile,
-          originalFileName: asset.originalFileName,
-          isVisible: false,
-          isReadOnly: true,
-          deviceAssetId: 'NONE',
-          deviceId: 'NONE',
-        });
-
-        await fs.writeFile(encodedFile, video.buffer);
-
-        await this.jobRepository.queue({ name: JobName.METADATA_EXTRACTION, data: { id: motionAsset.id } });
-      }
-
-      await this.assetRepository.save({ id: asset.id, livePhotoVideoId: motionAsset.id });
-
-      this.logger.debug(`Finished motion photo video extraction (${asset.id})`);
-    } catch (error: Error | any) {
-      this.logger.error(`Failed to extract live photo ${asset.originalPath}: ${error}`, error?.stack);
-    } finally {
-      await file?.close();
-    }
-  }
-
-  private async exifData(asset: AssetEntity): Promise<[ExifEntity, ImmichTags]> {
-    const readTaskOptions: ReadTaskOptions = {
-      ...DefaultReadTaskOptions,
-
-      defaultVideosToUTC: true,
-      backfillTimezones: true,
-      inferTimezoneFromDatestamps: true,
-      useMWG: true,
-      numericTags: DefaultReadTaskOptions.numericTags.concat(['FocalLength']),
-      geoTz: (lat: number, lon: number): string => geotz.find(lat, lon)[0],
-    };
-
-    const mediaTags = await exiftool
-      .read<ImmichTags>(asset.originalPath, undefined, readTaskOptions)
-      .catch((error: any) => {
-        this.logger.warn(`error reading exif data (${asset.id} at ${asset.originalPath}): ${error}`, error?.stack);
-        return null;
-      });
-
-    const sidecarTags = asset.sidecarPath
-      ? await exiftool.read<ImmichTags>(asset.sidecarPath, undefined, readTaskOptions).catch((error: any) => {
-          this.logger.warn(`error reading exif data (${asset.id} at ${asset.sidecarPath}): ${error}`, error?.stack);
-          return null;
-        })
-      : null;
-
-    const stats = await fs.stat(asset.originalPath);
-
-    const tags = { ...mediaTags, ...sidecarTags };
-
-    this.logger.verbose('Exif Tags', tags);
-
-    return [
-      <ExifEntity>{
-        // altitude: tags.GPSAltitude ?? null,
-        assetId: asset.id,
-        bitsPerSample: this.getBitsPerSample(tags),
-        colorspace: tags.ColorSpace ?? null,
-        dateTimeOriginal: exifDate(firstDateTime(tags)) ?? asset.fileCreatedAt,
-        exifImageHeight: validate(tags.ImageHeight),
-        exifImageWidth: validate(tags.ImageWidth),
-        exposureTime: tags.ExposureTime ?? null,
-        fileSizeInByte: stats.size,
-        fNumber: validate(tags.FNumber),
-        focalLength: validate(tags.FocalLength),
-        fps: validate(tags.VideoFrameRate),
-        iso: validate(tags.ISO),
-        latitude: validate(tags.GPSLatitude),
-        lensModel: tags.LensModel ?? null,
-        livePhotoCID: (asset.type === AssetType.VIDEO ? tags.ContentIdentifier : tags.MediaGroupUUID) ?? null,
-        longitude: validate(tags.GPSLongitude),
-        make: tags.Make ?? null,
-        model: tags.Model ?? null,
-        modifyDate: exifDate(tags.ModifyDate) ?? asset.fileModifiedAt,
-        orientation: validate(tags.Orientation)?.toString() ?? null,
-        profileDescription: tags.ProfileDescription || tags.ProfileName || null,
-        projectionType: tags.ProjectionType ? String(tags.ProjectionType).toUpperCase() : null,
-        timeZone: tags.tz,
-      },
-      tags,
-    ];
-  }
-
-  getBitsPerSample(tags: ImmichTags): number | null {
-    const bitDepthTags = [
-      tags.BitsPerSample,
-      tags.ComponentBitDepth,
-      tags.ImagePixelDepth,
-      tags.BitDepth,
-      tags.ColorBitDepth,
-      // `numericTags` doesn't parse values like '12 12 12'
-    ].map((tag) => (typeof tag === 'string' ? Number.parseInt(tag) : tag));
-
-    let bitsPerSample = bitDepthTags.find((tag) => typeof tag === 'number' && !Number.isNaN(tag)) ?? null;
-    if (bitsPerSample && bitsPerSample >= 24 && bitsPerSample % 3 === 0) {
-      bitsPerSample /= 3; // converts per-pixel bit depth to per-channel
-    }
-
-    return bitsPerSample;
-  }
-}

server/test/repositories/index.ts

@@ -10,6 +10,7 @@ export * from './job.repository.mock';
 export * from './library.repository.mock';
 export * from './machine-learning.repository.mock';
 export * from './media.repository.mock';
+export * from './metadata.repository.mock';
 export * from './partner.repository.mock';
 export * from './person.repository.mock';
 export * from './search.repository.mock';

server/test/repositories/metadata.repository.mock.ts

@@ -0,0 +1,10 @@
+import { IMetadataRepository } from '@app/domain';
+
+export const newMetadataRepositoryMock = (): jest.Mocked<IMetadataRepository> => {
+  return {
+    deleteCache: jest.fn(),
+    getExifTags: jest.fn(),
+    init: jest.fn(),
+    reverseGeocode: jest.fn(),
+  };
+};

server/test/repositories/storage.repository.mock.ts

@@ -4,6 +4,8 @@ export const newStorageRepositoryMock = (): jest.Mocked<IStorageRepository> => {
   return {
     createZipStream: jest.fn(),
     createReadStream: jest.fn(),
+    readFile: jest.fn(),
+    writeFile: jest.fn(),
     unlink: jest.fn(),
     unlinkDir: jest.fn().mockResolvedValue(true),
     removeEmptyDirs: jest.fn(),