feat(cli): concurrent upload (#7192)

* concurrent cli upload

* added concurrency flag, progress bar refinements

* no data property 🦀

* use lodash-es

* rebase

* linting

* typing

* album bug fixes

* dev dependency for lodash typing

* fixed not deleting assets if album isn't specified

* formatting

* fixed tests

* use `arrayContaining`

* add more checks

* assert updates existing assets

cli/package-lock.json

@@ -8,6 +8,9 @@
       "name": "@immich/cli",
       "version": "2.0.8",
       "license": "GNU Affero General Public License version 3",
+      "dependencies": {
+        "lodash-es": "^4.17.21"
+      },
       "bin": {
         "immich": "dist/index.js"
       },
@@ -16,6 +19,7 @@
         "@testcontainers/postgresql": "^10.7.1",
         "@types/byte-size": "^8.1.0",
         "@types/cli-progress": "^3.11.0",
+        "@types/lodash-es": "^4.17.12",
         "@types/mock-fs": "^4.13.1",
         "@types/node": "^20.3.1",
         "@typescript-eslint/eslint-plugin": "^7.0.0",
@@ -1296,6 +1300,21 @@
       "integrity": "sha512-5+fP8P8MFNC+AyZCDxrB2pkZFPGzqQWUzpSeuuVLvm8VMcorNYavBqoFcxK8bQz4Qsbn4oUEEem4wDLfcysGHA==",
       "dev": true
     },
+    "node_modules/@types/lodash": {
+      "version": "4.14.202",
+      "resolved": "https://registry.npmjs.org/@types/lodash/-/lodash-4.14.202.tgz",
+      "integrity": "sha512-OvlIYQK9tNneDlS0VN54LLd5uiPCBOp7gS5Z0f1mjoJYBrtStzgmJBxONW3U6OZqdtNzZPmn9BS/7WI7BFFcFQ==",
+      "dev": true
+    },
+    "node_modules/@types/lodash-es": {
+      "version": "4.17.12",
+      "resolved": "https://registry.npmjs.org/@types/lodash-es/-/lodash-es-4.17.12.tgz",
+      "integrity": "sha512-0NgftHUcV4v34VhXm8QBSftKVXtbkBG3ViCjs6+eJ5a6y6Mi/jiFGPc1sC7QK+9BFhWrURE3EOggmWaSxL9OzQ==",
+      "dev": true,
+      "dependencies": {
+        "@types/lodash": "*"
+      }
+    },
     "node_modules/@types/mock-fs": {
       "version": "4.13.4",
       "resolved": "https://registry.npmjs.org/@types/mock-fs/-/mock-fs-4.13.4.tgz",
@@ -3539,6 +3558,11 @@
         "url": "https://github.com/sponsors/sindresorhus"
       }
     },
+    "node_modules/lodash-es": {
+      "version": "4.17.21",
+      "resolved": "https://registry.npmjs.org/lodash-es/-/lodash-es-4.17.21.tgz",
+      "integrity": "sha512-mKnC+QJ9pWVzv+C4/U3rRsHapFfHvQFoFB92e52xeyGMcX6/OlIl78je1u8vePzYZSkkogMPJ2yjxxsb89cxyw=="
+    },
     "node_modules/lodash.defaults": {
       "version": "4.2.0",
       "resolved": "https://registry.npmjs.org/lodash.defaults/-/lodash.defaults-4.2.0.tgz",
@@ -6432,6 +6456,21 @@
       "integrity": "sha512-5+fP8P8MFNC+AyZCDxrB2pkZFPGzqQWUzpSeuuVLvm8VMcorNYavBqoFcxK8bQz4Qsbn4oUEEem4wDLfcysGHA==",
       "dev": true
     },
+    "@types/lodash": {
+      "version": "4.14.202",
+      "resolved": "https://registry.npmjs.org/@types/lodash/-/lodash-4.14.202.tgz",
+      "integrity": "sha512-OvlIYQK9tNneDlS0VN54LLd5uiPCBOp7gS5Z0f1mjoJYBrtStzgmJBxONW3U6OZqdtNzZPmn9BS/7WI7BFFcFQ==",
+      "dev": true
+    },
+    "@types/lodash-es": {
+      "version": "4.17.12",
+      "resolved": "https://registry.npmjs.org/@types/lodash-es/-/lodash-es-4.17.12.tgz",
+      "integrity": "sha512-0NgftHUcV4v34VhXm8QBSftKVXtbkBG3ViCjs6+eJ5a6y6Mi/jiFGPc1sC7QK+9BFhWrURE3EOggmWaSxL9OzQ==",
+      "dev": true,
+      "requires": {
+        "@types/lodash": "*"
+      }
+    },
     "@types/mock-fs": {
       "version": "4.13.4",
       "resolved": "https://registry.npmjs.org/@types/mock-fs/-/mock-fs-4.13.4.tgz",
@@ -8080,6 +8119,11 @@
         "p-locate": "^5.0.0"
       }
     },
+    "lodash-es": {
+      "version": "4.17.21",
+      "resolved": "https://registry.npmjs.org/lodash-es/-/lodash-es-4.17.21.tgz",
+      "integrity": "sha512-mKnC+QJ9pWVzv+C4/U3rRsHapFfHvQFoFB92e52xeyGMcX6/OlIl78je1u8vePzYZSkkogMPJ2yjxxsb89cxyw=="
+    },
     "lodash.defaults": {
       "version": "4.2.0",
       "resolved": "https://registry.npmjs.org/lodash.defaults/-/lodash.defaults-4.2.0.tgz",

cli/package.json

@@ -17,6 +17,7 @@
     "@testcontainers/postgresql": "^10.7.1",
     "@types/byte-size": "^8.1.0",
     "@types/cli-progress": "^3.11.0",
+    "@types/lodash-es": "^4.17.12",
     "@types/mock-fs": "^4.13.1",
     "@types/node": "^20.3.1",
     "@typescript-eslint/eslint-plugin": "^7.0.0",
@@ -56,5 +57,8 @@
   },
   "engines": {
     "node": ">=20.0.0"
+  },
+  "dependencies": {
+    "lodash-es": "^4.17.21"
   }
 }

cli/src/commands/upload.command.ts

@@ -1,5 +1,7 @@
+import { AssetBulkUploadCheckResult } from '@immich/sdk';
 import byteSize from 'byte-size';
 import cliProgress from 'cli-progress';
+import { chunk, zip } from 'lodash-es';
 import { createHash } from 'node:crypto';
 import fs, { createReadStream } from 'node:fs';
 import { access, constants, stat, unlink } from 'node:fs/promises';
@@ -9,15 +11,23 @@ import { ImmichApi } from 'src/services/api.service';
 import { CrawlService } from '../services/crawl.service';
 import { BaseCommand } from './base-command';
 
+const zipDefined = zip as <T, U>(a: T[], b: U[]) => [T, U][];
+
+enum CheckResponseStatus {
+  ACCEPT = 'accept',
+  REJECT = 'reject',
+  DUPLICATE = 'duplicate',
+}
+
 class Asset {
   readonly path: string;
-  readonly deviceId!: string;
 
+  id?: string;
   deviceAssetId?: string;
   fileCreatedAt?: Date;
   fileModifiedAt?: Date;
   sidecarPath?: string;
-  fileSize!: number;
+  fileSize?: number;
   albumName?: string;
 
   constructor(path: string) {
@@ -105,17 +115,141 @@ export class UploadOptionsDto {
   album? = false;
   albumName? = '';
   includeHidden? = false;
+  concurrency? = 4;
 }
 
 export class UploadCommand extends BaseCommand {
-  uploadLength!: number;
+  api!: ImmichApi;
 
   public async run(paths: string[], options: UploadOptionsDto): Promise<void> {
-    const api = await this.connect();
+    this.api = await this.connect();
 
-    const formatResponse = await api.getSupportedMediaTypes();
+    console.log('Crawling for assets...');
-    const crawlService = new CrawlService(formatResponse.image, formatResponse.video);
+    const files = await this.getFiles(paths, options);
 
+    if (files.length === 0) {
+      console.log('No assets found, exiting');
+      return;
+    }
+
+    const assetsToCheck = files.map((path) => new Asset(path));
+
+    const { newAssets, duplicateAssets } = await this.checkAssets(assetsToCheck, options.concurrency ?? 4);
+
+    const totalSizeUploaded = await this.upload(newAssets, options);
+    const messageStart = options.dryRun ? 'Would have' : 'Successfully';
+    if (newAssets.length === 0) {
+      console.log('All assets were already uploaded, nothing to do.');
+    } else {
+      console.log(
+        `${messageStart} uploaded ${newAssets.length} asset${newAssets.length === 1 ? '' : 's'} (${byteSize(totalSizeUploaded)})`,
+      );
+    }
+
+    if (options.album || options.albumName) {
+      const { createdAlbumCount, updatedAssetCount } = await this.updateAlbums(
+        [...newAssets, ...duplicateAssets],
+        options,
+      );
+      console.log(`${messageStart} created ${createdAlbumCount} new album${createdAlbumCount === 1 ? '' : 's'}`);
+      console.log(`${messageStart} updated ${updatedAssetCount} asset${updatedAssetCount === 1 ? '' : 's'}`);
+    }
+
+    if (!options.delete) {
+      return;
+    }
+
+    if (options.dryRun) {
+      console.log(`Would now have deleted assets, but skipped due to dry run`);
+      return;
+    }
+
+    console.log('Deleting assets that have been uploaded...');
+
+    await this.deleteAssets(newAssets, options);
+  }
+
+  public async checkAssets(
+    assetsToCheck: Asset[],
+    concurrency: number,
+  ): Promise<{ newAssets: Asset[]; duplicateAssets: Asset[]; rejectedAssets: Asset[] }> {
+    for (const assets of chunk(assetsToCheck, concurrency)) {
+      await Promise.all(assets.map((asset: Asset) => asset.prepare()));
+    }
+
+    const checkProgress = new cliProgress.SingleBar(
+      { format: 'Checking assets | {bar} | {percentage}% | ETA: {eta}s | {value}/{total} assets' },
+      cliProgress.Presets.shades_classic,
+    );
+    checkProgress.start(assetsToCheck.length, 0);
+
+    const newAssets = [];
+    const duplicateAssets = [];
+    const rejectedAssets = [];
+    try {
+      for (const assets of chunk(assetsToCheck, concurrency)) {
+        const checkedAssets = await this.getStatus(assets);
+        for (const checked of checkedAssets) {
+          if (checked.status === CheckResponseStatus.ACCEPT) {
+            newAssets.push(checked.asset);
+          } else if (checked.status === CheckResponseStatus.DUPLICATE) {
+            duplicateAssets.push(checked.asset);
+          } else {
+            rejectedAssets.push(checked.asset);
+          }
+          checkProgress.increment();
+        }
+      }
+    } finally {
+      checkProgress.stop();
+    }
+
+    return { newAssets, duplicateAssets, rejectedAssets };
+  }
+
+  public async upload(assetsToUpload: Asset[], options: UploadOptionsDto): Promise<number> {
+    let totalSize = 0;
+
+    // Compute total size first
+    for (const asset of assetsToUpload) {
+      totalSize += asset.fileSize ?? 0;
+    }
+
+    if (options.dryRun) {
+      return totalSize;
+    }
+
+    const uploadProgress = new cliProgress.SingleBar(
+      {
+        format: 'Uploading assets | {bar} | {percentage}% | ETA: {eta_formatted} | {value_formatted}/{total_formatted}',
+      },
+      cliProgress.Presets.shades_classic,
+    );
+    uploadProgress.start(totalSize, 0);
+    uploadProgress.update({ value_formatted: 0, total_formatted: byteSize(totalSize) });
+
+    let totalSizeUploaded = 0;
+    try {
+      for (const assets of chunk(assetsToUpload, options.concurrency)) {
+        const ids = await this.uploadAssets(assets);
+        for (const [asset, id] of zipDefined(assets, ids)) {
+          asset.id = id;
+          if (asset.fileSize) {
+            totalSizeUploaded += asset.fileSize ?? 0;
+          } else {
+            console.log(`Could not determine file size for ${asset.path}`);
+          }
+        }
+        uploadProgress.update(totalSizeUploaded, { value_formatted: byteSize(totalSizeUploaded) });
+      }
+    } finally {
+      uploadProgress.stop();
+    }
+
+    return totalSizeUploaded;
+  }
+
+  public async getFiles(paths: string[], options: UploadOptionsDto): Promise<string[]> {
     const inputFiles: string[] = [];
     for (const pathArgument of paths) {
       const fileStat = await fs.promises.lstat(pathArgument);
@@ -124,151 +258,187 @@ export class UploadCommand extends BaseCommand {
       }
     }
 
-    const files: string[] = await crawlService.crawl({
+    const files: string[] = await this.crawl(paths, options);
+    files.push(...inputFiles);
+    return files;
+  }
+
+  public async getAlbums(): Promise<Map<string, string>> {
+    const existingAlbums = await this.api.getAllAlbums();
+
+    const albumMapping = new Map<string, string>();
+    for (const album of existingAlbums) {
+      albumMapping.set(album.albumName, album.id);
+    }
+
+    return albumMapping;
+  }
+
+  public async updateAlbums(
+    assets: Asset[],
+    options: UploadOptionsDto,
+  ): Promise<{ createdAlbumCount: number; updatedAssetCount: number }> {
+    if (options.albumName) {
+      for (const asset of assets) {
+        asset.albumName = options.albumName;
+      }
+    }
+
+    const existingAlbums = await this.getAlbums();
+    const assetsToUpdate = assets.filter(
+      (asset): asset is Asset & { albumName: string; id: string } => !!(asset.albumName && asset.id),
+    );
+
+    const newAlbumsSet: Set<string> = new Set();
+    for (const asset of assetsToUpdate) {
+      if (!existingAlbums.has(asset.albumName)) {
+        newAlbumsSet.add(asset.albumName);
+      }
+    }
+
+    const newAlbums = [...newAlbumsSet];
+
+    if (options.dryRun) {
+      return { createdAlbumCount: newAlbums.length, updatedAssetCount: assetsToUpdate.length };
+    }
+
+    const albumCreationProgress = new cliProgress.SingleBar(
+      {
+        format: 'Creating albums | {bar} | {percentage}% | ETA: {eta}s | {value}/{total} albums',
+      },
+      cliProgress.Presets.shades_classic,
+    );
+    albumCreationProgress.start(newAlbums.length, 0);
+
+    try {
+      for (const albumNames of chunk(newAlbums, options.concurrency)) {
+        const newAlbumIds = await Promise.all(
+          albumNames.map((albumName: string) => this.api.createAlbum({ albumName }).then((r) => r.id)),
+        );
+
+        for (const [albumName, albumId] of zipDefined(albumNames, newAlbumIds)) {
+          existingAlbums.set(albumName, albumId);
+        }
+
+        albumCreationProgress.increment(albumNames.length);
+      }
+    } finally {
+      albumCreationProgress.stop();
+    }
+
+    const albumToAssets = new Map<string, string[]>();
+    for (const asset of assetsToUpdate) {
+      const albumId = existingAlbums.get(asset.albumName);
+      if (albumId) {
+        if (!albumToAssets.has(albumId)) {
+          albumToAssets.set(albumId, []);
+        }
+        albumToAssets.get(albumId)?.push(asset.id);
+      }
+    }
+
+    const albumUpdateProgress = new cliProgress.SingleBar(
+      {
+        format: 'Adding assets to albums | {bar} | {percentage}% | ETA: {eta}s | {value}/{total} assets',
+      },
+      cliProgress.Presets.shades_classic,
+    );
+    albumUpdateProgress.start(assetsToUpdate.length, 0);
+
+    try {
+      for (const [albumId, assets] of albumToAssets.entries()) {
+        for (const assetBatch of chunk(assets, Math.min(1000 * (options.concurrency ?? 4), 65_000))) {
+          await this.api.addAssetsToAlbum(albumId, { ids: assetBatch });
+          albumUpdateProgress.increment(assetBatch.length);
+        }
+      }
+    } finally {
+      albumUpdateProgress.stop();
+    }
+
+    return { createdAlbumCount: newAlbums.length, updatedAssetCount: assetsToUpdate.length };
+  }
+
+  public async deleteAssets(assets: Asset[], options: UploadOptionsDto): Promise<void> {
+    const deletionProgress = new cliProgress.SingleBar(
+      {
+        format: 'Deleting local assets | {bar} | {percentage}% | ETA: {eta}s | {value}/{total} assets',
+      },
+      cliProgress.Presets.shades_classic,
+    );
+    deletionProgress.start(assets.length, 0);
+
+    try {
+      for (const assetBatch of chunk(assets, options.concurrency)) {
+        await Promise.all(assetBatch.map((asset: Asset) => asset.delete()));
+        deletionProgress.update(assetBatch.length);
+      }
+    } finally {
+      deletionProgress.stop();
+    }
+  }
+
+  private async getStatus(assets: Asset[]): Promise<{ asset: Asset; status: CheckResponseStatus }[]> {
+    const checkResponse = await this.checkHashes(assets);
+
+    const responses = [];
+    for (const [check, asset] of zipDefined(checkResponse, assets)) {
+      if (check.assetId) {
+        asset.id = check.assetId;
+      }
+
+      if (check.action === 'accept') {
+        responses.push({ asset, status: CheckResponseStatus.ACCEPT });
+      } else if (check.reason === 'duplicate') {
+        responses.push({ asset, status: CheckResponseStatus.DUPLICATE });
+      } else {
+        responses.push({ asset, status: CheckResponseStatus.REJECT });
+      }
+    }
+
+    return responses;
+  }
+
+  private async checkHashes(assetsToCheck: Asset[]): Promise<AssetBulkUploadCheckResult[]> {
+    const checksums = await Promise.all(assetsToCheck.map((asset) => asset.hash()));
+    const assetBulkUploadCheckDto = {
+      assets: zipDefined(assetsToCheck, checksums).map(([asset, checksum]) => ({ id: asset.path, checksum })),
+    };
+    const checkResponse = await this.api.checkBulkUpload(assetBulkUploadCheckDto);
+    return checkResponse.results;
+  }
+
+  private async uploadAssets(assets: Asset[]): Promise<string[]> {
+    const fileRequests = await Promise.all(assets.map((asset) => asset.getUploadFormData()));
+    return Promise.all(fileRequests.map((request) => this.uploadAsset(request).then((response) => response.id)));
+  }
+
+  private async crawl(paths: string[], options: UploadOptionsDto): Promise<string[]> {
+    const formatResponse = await this.api.getSupportedMediaTypes();
+    const crawlService = new CrawlService(formatResponse.image, formatResponse.video);
+
+    return crawlService.crawl({
       pathsToCrawl: paths,
       recursive: options.recursive,
       exclusionPatterns: options.exclusionPatterns,
       includeHidden: options.includeHidden,
     });
-
-    files.push(...inputFiles);
-
-    if (files.length === 0) {
-      console.log('No assets found, exiting');
-      return;
-    }
-
-    const assetsToUpload = files.map((path) => new Asset(path));
-
-    const uploadProgress = new cliProgress.SingleBar(
-      {
-        format: '{bar} | {percentage}% | ETA: {eta_formatted} | {value_formatted}/{total_formatted}: {filename}',
-      },
-      cliProgress.Presets.shades_classic,
-    );
-
-    let totalSize = 0;
-    let sizeSoFar = 0;
-
-    let totalSizeUploaded = 0;
-    let uploadCounter = 0;
-
-    for (const asset of assetsToUpload) {
-      // Compute total size first
-      await asset.prepare();
-      totalSize += asset.fileSize;
-
-      if (options.albumName) {
-        asset.albumName = options.albumName;
-      }
-    }
-
-    const existingAlbums = await api.getAllAlbums();
-
-    uploadProgress.start(totalSize, 0);
-    uploadProgress.update({ value_formatted: 0, total_formatted: byteSize(totalSize) });
-
-    try {
-      for (const asset of assetsToUpload) {
-        uploadProgress.update({
-          filename: asset.path,
-        });
-
-        let skipUpload = false;
-
-        let skipAsset = false;
-        let existingAssetId: string | undefined = undefined;
-
-        if (!options.skipHash) {
-          const assetBulkUploadCheckDto = { assets: [{ id: asset.path, checksum: await asset.hash() }] };
-
-          const checkResponse = await api.checkBulkUpload(assetBulkUploadCheckDto);
-
-          skipUpload = checkResponse.results[0].action === 'reject';
-
-          const isDuplicate = checkResponse.results[0].reason === 'duplicate';
-          if (isDuplicate) {
-            existingAssetId = checkResponse.results[0].assetId;
-          }
-
-          skipAsset = skipUpload && !isDuplicate;
-        }
-
-        if (!skipAsset && !options.dryRun) {
-          if (!skipUpload) {
-            const formData = await asset.getUploadFormData();
-            const response = await this.uploadAsset(api, formData);
-            const json = await response.json();
-            existingAssetId = json.id;
-            uploadCounter++;
-            totalSizeUploaded += asset.fileSize;
-          }
-
-          if ((options.album || options.albumName) && asset.albumName !== undefined) {
-            let album = existingAlbums.find((album) => album.albumName === asset.albumName);
-            if (!album) {
-              const response = await api.createAlbum({ albumName: asset.albumName });
-              album = response;
-              existingAlbums.push(album);
-            }
-
-            if (existingAssetId) {
-              await api.addAssetsToAlbum(album.id, {
-                ids: [existingAssetId],
-              });
-            }
-          }
-        }
-
-        sizeSoFar += asset.fileSize;
-
-        uploadProgress.update(sizeSoFar, { value_formatted: byteSize(sizeSoFar) });
-      }
-    } finally {
-      uploadProgress.stop();
-    }
-
-    const messageStart = options.dryRun ? 'Would have' : 'Successfully';
-
-    if (uploadCounter === 0) {
-      console.log('All assets were already uploaded, nothing to do.');
-    } else {
-      console.log(`${messageStart} uploaded ${uploadCounter} assets (${byteSize(totalSizeUploaded)})`);
-    }
-    if (options.delete) {
-      if (options.dryRun) {
-        console.log(`Would now have deleted assets, but skipped due to dry run`);
-      } else {
-        console.log('Deleting assets that have been uploaded...');
-        const deletionProgress = new cliProgress.SingleBar(cliProgress.Presets.shades_classic);
-        deletionProgress.start(files.length, 0);
-
-        for (const asset of assetsToUpload) {
-          if (!options.dryRun) {
-            await asset.delete();
-          }
-          deletionProgress.increment();
-        }
-        deletionProgress.stop();
-        console.log('Deletion complete');
-      }
-    }
   }
 
-  private async uploadAsset(api: ImmichApi, data: FormData): Promise<Response> {
+  private async uploadAsset(data: FormData): Promise<{ id: string }> {
-    const url = api.instanceUrl + '/asset/upload';
+    const url = this.api.instanceUrl + '/asset/upload';
 
     const response = await fetch(url, {
       method: 'post',
       redirect: 'error',
       headers: {
-        'x-api-key': api.apiKey,
+        'x-api-key': this.api.apiKey,
       },
       body: data,
     });
     if (response.status !== 200 && response.status !== 201) {
       throw new Error(await response.text());
     }
-    return response;
+    return response.json();
   }
 }

cli/src/index.ts

@@ -43,6 +43,11 @@ program
       .env('IMMICH_DRY_RUN')
       .default(false),
   )
+  .addOption(
+    new Option('-c, --concurrency', 'Number of assets to upload at the same time')
+      .env('IMMICH_UPLOAD_CONCURRENCY')
+      .default(4),
+  )
   .addOption(new Option('--delete', 'Delete local assets after upload').env('IMMICH_DELETE_ASSETS'))
   .argument('[paths...]', 'One or more paths to assets to be uploaded')
   .action(async (paths, options) => {

e2e/src/cli/specs/upload.e2e-spec.ts

@@ -8,6 +8,7 @@ import {
   testAssetDir,
 } from 'src/utils';
 import { beforeAll, beforeEach, describe, expect, it } from 'vitest';
+import { mkdir, readdir, rm, symlink } from 'fs/promises';
 
 describe(`immich upload`, () => {
   let key: string;
@@ -29,9 +30,11 @@ describe(`immich upload`, () => {
         '--recursive',
       ]);
       expect(stderr).toBe('');
-      expect(stdout.split('\n')).toEqual([
+      expect(stdout.split('\n')).toEqual(
-        expect.stringContaining('Successfully uploaded 9 assets'),
+        expect.arrayContaining([
-      ]);
+          expect.stringContaining('Successfully uploaded 9 assets'),
+        ])
+      );
       expect(exitCode).toBe(0);
 
       const assets = await getAllAssets({}, { headers: asKeyAuth(key) });
@@ -47,9 +50,13 @@ describe(`immich upload`, () => {
         '--recursive',
         '--album',
       ]);
-      expect(stdout.split('\n')).toEqual([
+      expect(stdout.split('\n')).toEqual(
-        expect.stringContaining('Successfully uploaded 9 assets'),
+        expect.arrayContaining([
-      ]);
+          expect.stringContaining('Successfully uploaded 9 assets'),
+          expect.stringContaining('Successfully created 1 new album'),
+          expect.stringContaining('Successfully updated 9 assets'),
+        ])
+      );
       expect(stderr).toBe('');
       expect(exitCode).toBe(0);
 
@@ -67,9 +74,11 @@ describe(`immich upload`, () => {
         `${testAssetDir}/albums/nature/`,
         '--recursive',
       ]);
-      expect(response1.stdout.split('\n')).toEqual([
+      expect(response1.stdout.split('\n')).toEqual(
-        expect.stringContaining('Successfully uploaded 9 assets'),
+        expect.arrayContaining([
-      ]);
+          expect.stringContaining('Successfully uploaded 9 assets'),
+        ])
+      );
       expect(response1.stderr).toBe('');
       expect(response1.exitCode).toBe(0);
 
@@ -85,11 +94,14 @@ describe(`immich upload`, () => {
         '--recursive',
         '--album',
       ]);
-      expect(response2.stdout.split('\n')).toEqual([
+      expect(response2.stdout.split('\n')).toEqual(
-        expect.stringContaining(
+        expect.arrayContaining([
-          'All assets were already uploaded, nothing to do.'
+          expect.stringContaining(
-        ),
+            'All assets were already uploaded, nothing to do.'
-      ]);
+          ),
+          expect.stringContaining('Successfully updated 9 assets'),
+        ])
+      );
       expect(response2.stderr).toBe('');
       expect(response2.exitCode).toBe(0);
 
@@ -110,9 +122,13 @@ describe(`immich upload`, () => {
         '--recursive',
         '--album-name=e2e',
       ]);
-      expect(stdout.split('\n')).toEqual([
+      expect(stdout.split('\n')).toEqual(
-        expect.stringContaining('Successfully uploaded 9 assets'),
+        expect.arrayContaining([
-      ]);
+          expect.stringContaining('Successfully uploaded 9 assets'),
+          expect.stringContaining('Successfully created 1 new album'),
+          expect.stringContaining('Successfully updated 9 assets'),
+        ])
+      );
       expect(stderr).toBe('');
       expect(exitCode).toBe(0);
 
@@ -124,4 +140,39 @@ describe(`immich upload`, () => {
       expect(albums[0].albumName).toBe('e2e');
     });
   });
+
+  describe('immich upload --delete', () => {
+    it('should delete local files if specified', async () => {
+      await mkdir(`/tmp/albums/nature`, { recursive: true });
+      const filesToLink = await readdir(`${testAssetDir}/albums/nature`);
+      for (const file of filesToLink) {
+        await symlink(
+          `${testAssetDir}/albums/nature/${file}`,
+          `/tmp/albums/nature/${file}`
+        );
+      }
+
+      const { stderr, stdout, exitCode } = await immichCli([
+        'upload',
+        `/tmp/albums/nature`,
+        '--delete',
+      ]);
+
+      const files = await readdir(`/tmp/albums/nature`);
+      await rm(`/tmp/albums/nature`, { recursive: true });
+      expect(files).toEqual([]);
+
+      expect(stdout.split('\n')).toEqual(
+        expect.arrayContaining([
+          expect.stringContaining('Successfully uploaded 9 assets'),
+          expect.stringContaining('Deleting assets that have been uploaded'),
+        ])
+      );
+      expect(stderr).toBe('');
+      expect(exitCode).toBe(0);
+
+      const assets = await getAllAssets({}, { headers: asKeyAuth(key) });
+      expect(assets.length).toBe(9);
+    });
+  });
 });