feat(server): de-duplication (#557)

* feat(server): remove un-used deviceAssetId cols.

* feat(server): return 409 if asset is duplicated

* feat(server): replace old unique constaint

* feat(server): strip deviceId in file path

* feat(server): skip duplicate asset

* chore(server): revert changes

* fix(server): asset test spec

* fix(server): checksum generation for uploaded assets

* fix(server): make sure generation queue run after migraion

* feat(server): remove temp file

* chore(server): remove dead code

server/apps/immich/src/api-v1/asset/asset-repository.ts

@@ -26,6 +26,7 @@ export interface IAssetRepository {
   getSearchPropertiesByUserId(userId: string): Promise<SearchPropertiesDto[]>;
   getAssetCountByTimeBucket(userId: string, timeBucket: TimeGroupEnum): Promise<AssetCountByTimeBucket[]>;
   getAssetByTimeBucket(userId: string, getAssetByTimeBucketDto: GetAssetByTimeBucketDto): Promise<AssetEntity[]>;
+  getAssetByChecksum(userId: string, checksum: Buffer): Promise<AssetEntity>;
 }
 
 export const ASSET_REPOSITORY = 'ASSET_REPOSITORY';
@@ -208,4 +209,20 @@ export class AssetRepository implements IAssetRepository {
 
     return res;
   }
+
+  /**
+   * Get asset by checksum on the database
+   * @param userId 
+   * @param checksum 
+   * 
+   */
+  getAssetByChecksum(userId: string, checksum: Buffer): Promise<AssetEntity> {
+    return this.assetRepository.findOneOrFail({
+      where: {
+        userId,
+        checksum
+      },
+      relations: ['exifInfo'],
+    });
+  }
 }

server/apps/immich/src/api-v1/asset/asset.service.spec.ts

@@ -60,6 +60,7 @@ describe('AssetService', () => {
       getLocationsByUserId: jest.fn(),
       getSearchPropertiesByUserId: jest.fn(),
       getAssetByTimeBucket: jest.fn(),
+      getAssetByChecksum: jest.fn(),
     };
 
     sui = new AssetService(assetRepositoryMock, a);

server/apps/immich/src/api-v1/asset/asset.service.ts

@@ -10,7 +10,7 @@ import {
 } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 import { createHash } from 'node:crypto';
-import { Repository } from 'typeorm';
+import { QueryFailedError, Repository } from 'typeorm';
 import { AuthUserDto } from '../../decorators/auth-user.decorator';
 import { AssetEntity, AssetType } from '@app/database/entities/asset.entity';
 import { constants, createReadStream, ReadStream, stat } from 'fs';
@@ -55,15 +55,29 @@ export class AssetService {
     mimeType: string,
   ): Promise<AssetEntity> {
     const checksum = await this.calculateChecksum(originalPath);
-    const assetEntity = await this._assetRepository.create(
-      createAssetDto,
-      authUser.id,
-      originalPath,
-      mimeType,
-      checksum,
-    );
 
-    return assetEntity;
+    try {
+      const assetEntity = await this._assetRepository.create(
+        createAssetDto,
+        authUser.id,
+        originalPath,
+        mimeType,
+        checksum,
+      );
+
+      return assetEntity;
+    } catch (err) {
+      if (err instanceof QueryFailedError && (err as any).constraint === 'UQ_userid_checksum') {
+        const [assetEntity, _] = await Promise.all([
+          this._assetRepository.getAssetByChecksum(authUser.id, checksum),
+          fs.unlink(originalPath)
+        ]);
+
+        return assetEntity;
+      }
+
+      throw err;
+    }
   }
 
   public async getUserAssetsByDeviceId(authUser: AuthUserDto, deviceId: string) {

server/apps/immich/src/config/asset-upload.config.ts

@@ -3,7 +3,7 @@ import { HttpException, HttpStatus } from '@nestjs/common';
 import { MulterOptions } from '@nestjs/platform-express/multer/interfaces/multer-options.interface';
 import { existsSync, mkdirSync } from 'fs';
 import { diskStorage } from 'multer';
-import { extname } from 'path';
+import { extname, join } from 'path';
 import { Request } from 'express';
 import { randomUUID } from 'crypto';
 
@@ -29,7 +29,7 @@ export const assetUploadOption: MulterOptions = {
         return;
       }
 
-      const originalUploadFolder = `${basePath}/${req.user.id}/original/${req.body['deviceId']}`;
+      const originalUploadFolder = join(basePath, req.user.id, 'original', req.body['deviceId']);
 
       if (!existsSync(originalUploadFolder)) {
         mkdirSync(originalUploadFolder, { recursive: true });

server/apps/microservices/src/microservices.service.ts

@@ -12,6 +12,8 @@ export class MicroservicesService implements OnModuleInit {
   ) {}
 
   async onModuleInit() {
-    await this.generateChecksumQueue.add({}, { jobId: randomUUID() },);
+    await this.generateChecksumQueue.add({}, {
+      jobId: randomUUID(), delay: 10000 // wait for migration
+    });
   }
 }

server/apps/microservices/src/processors/generate-checksum.processor.ts

@@ -19,14 +19,12 @@ export class GenerateChecksumProcessor {
   async generateChecksum() {
     let hasNext = true;
     let pageSize = 200;
-    let offset = 0;
 
     while (hasNext) {
       const assets = await this.assetRepository.find({
         where: {
           checksum: IsNull()
         },
-        skip: offset,
         take: pageSize,
       });
 
@@ -43,8 +41,6 @@ export class GenerateChecksumProcessor {
 
         if (assets.length < pageSize) {
           hasNext = false;
-        } else {
-          offset += pageSize;
         }
       }
     }

server/libs/database/src/entities/asset.entity.ts

@@ -3,7 +3,7 @@ import { ExifEntity } from './exif.entity';
 import { SmartInfoEntity } from './smart-info.entity';
 
 @Entity('assets')
-@Unique(['deviceAssetId', 'userId', 'deviceId'])
+@Unique('UQ_userid_checksum', ['userId', 'checksum'])
 export class AssetEntity {
   @PrimaryGeneratedColumn('uuid')
   id!: string;

server/libs/database/src/migrations/1661881837496-AddAssetChecksum.ts

@@ -1,4 +1,4 @@
-import { MigrationInterface, QueryRunner } from "typeorm";
+import { MigrationInterface, QueryRunner } from 'typeorm';
 
 export class AddAssetChecksum1661881837496 implements MigrationInterface {
   name = 'AddAssetChecksum1661881837496'

server/libs/database/src/migrations/1661971370662-UpdateAssetTableWithNewUniqueConstraint.ts

@@ -0,0 +1,16 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class UpdateAssetTableWithNewUniqueConstraint1661971370662 implements MigrationInterface {
+  name = 'UpdateAssetTableWithNewUniqueConstraint1661971370662'
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`ALTER TABLE "assets" DROP CONSTRAINT "UQ_b599ab0bd9574958acb0b30a90e"`);
+    await queryRunner.query(`ALTER TABLE "assets" ADD CONSTRAINT "UQ_userid_checksum" UNIQUE ("userId", "checksum")`);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`ALTER TABLE "assets" DROP CONSTRAINT "UQ_userid_checksum"`);
+    await queryRunner.query(`ALTER TABLE "assets" ADD CONSTRAINT "UQ_b599ab0bd9574958acb0b30a90e" UNIQUE ("deviceAssetId", "userId", "deviceId")`);
+  }
+
+}