refactor: library watching (#7071)

2025-12-22 01:11:20 +03:00 · 2024-02-13 08:48:47 -05:00
parent 3d7a7bcb7a
commit b648025e2f
9 changed files with 143 additions and 248 deletions
--- a/server/src/domain/library/library.service.spec.ts
+++ b/server/src/domain/library/library.service.spec.ts
@@ -1,11 +1,11 @@
 import { AssetType, LibraryType, SystemConfig, SystemConfigKey, UserEntity } from '@app/infra/entities';
 import { BadRequestException } from '@nestjs/common';
-
 import {
  IAccessRepositoryMock,
  assetStub,
  authStub,
  libraryStub,
+  makeMockWatcher,
  newAccessRepositoryMock,
  newAssetRepositoryMock,
  newCryptoRepositoryMock,
@@ -17,8 +17,6 @@ import {
  systemConfigStub,
  userStub,
 } from '@test';
-
-import { newFSWatcherMock } from '@test/mocks';
 import { Stats } from 'node:fs';
 import { ILibraryFileJob, ILibraryRefreshJob, JobName } from '../job';
 import {
@@ -128,16 +126,6 @@ describe(LibraryService.name, () => {
        }
      });

-      const mockWatcher = newFSWatcherMock();
-
-      mockWatcher.on.mockImplementation((event, callback) => {
-        if (event === 'ready') {
-          callback();
-        }
-      });
-
-      storageMock.watch.mockReturnValue(mockWatcher);
-
      await sut.init();

      expect(storageMock.watch.mock.calls).toEqual(
@@ -718,21 +706,13 @@ describe(LibraryService.name, () => {

      configMock.load.mockResolvedValue(systemConfigStub.libraryWatchEnabled);

-      const mockWatcher = newFSWatcherMock();
-
-      mockWatcher.on.mockImplementation((event, callback) => {
-        if (event === 'ready') {
-          callback();
-        }
-      });
-
-      storageMock.watch.mockReturnValue(mockWatcher);
+      const mockClose = jest.fn();
+      storageMock.watch.mockImplementation(makeMockWatcher({ close: mockClose }));

      await sut.init();
-
      await sut.delete(authStub.admin, libraryStub.externalLibraryWithImportPaths1.id);

-      expect(mockWatcher.close).toHaveBeenCalled();
+      expect(mockClose).toHaveBeenCalled();
    });
  });

@@ -940,16 +920,6 @@ describe(LibraryService.name, () => {
        libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
        libraryMock.getAll.mockResolvedValue([]);

-        const mockWatcher = newFSWatcherMock();
-
-        mockWatcher.on.mockImplementation((event, callback) => {
-          if (event === 'ready') {
-            callback();
-          }
-        });
-
-        storageMock.watch.mockReturnValue(mockWatcher);
-
        await sut.init();
        await sut.create(authStub.admin, {
          type: LibraryType.EXTERNAL,
@@ -959,6 +929,7 @@ describe(LibraryService.name, () => {
        expect(storageMock.watch).toHaveBeenCalledWith(
          libraryStub.externalLibraryWithImportPaths1.importPaths,
          expect.anything(),
+          expect.anything(),
        );
      });

@@ -1133,16 +1104,6 @@ describe(LibraryService.name, () => {
      libraryMock.update.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
      libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);

-      const mockWatcher = newFSWatcherMock();
-
-      mockWatcher.on.mockImplementation((event, callback) => {
-        if (event === 'ready') {
-          callback();
-        }
-      });
-
-      storageMock.watch.mockReturnValue(mockWatcher);
-
      await expect(sut.update(authStub.admin, authStub.admin.user.id, { importPaths: ['/foo'] })).resolves.toEqual(
        mapLibrary(libraryStub.externalLibraryWithImportPaths1),
      );
@@ -1155,6 +1116,7 @@ describe(LibraryService.name, () => {
      expect(storageMock.watch).toHaveBeenCalledWith(
        libraryStub.externalLibraryWithImportPaths1.importPaths,
        expect.anything(),
+        expect.anything(),
      );
    });

@@ -1163,16 +1125,6 @@ describe(LibraryService.name, () => {
      configMock.load.mockResolvedValue(systemConfigStub.libraryWatchEnabled);
      libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);

-      const mockWatcher = newFSWatcherMock();
-
-      mockWatcher.on.mockImplementation((event, callback) => {
-        if (event === 'ready') {
-          callback();
-        }
-      });
-
-      storageMock.watch.mockReturnValue(mockWatcher);
-
      await expect(sut.update(authStub.admin, authStub.admin.user.id, { exclusionPatterns: ['bar'] })).resolves.toEqual(
        mapLibrary(libraryStub.externalLibraryWithImportPaths1),
      );
@@ -1182,7 +1134,11 @@ describe(LibraryService.name, () => {
          id: authStub.admin.user.id,
        }),
      );
-      expect(storageMock.watch).toHaveBeenCalledWith(expect.arrayContaining([expect.any(String)]), expect.anything());
+      expect(storageMock.watch).toHaveBeenCalledWith(
+        expect.arrayContaining([expect.any(String)]),
+        expect.anything(),
+        expect.anything(),
+      );
    });
  });

@@ -1204,56 +1160,35 @@ describe(LibraryService.name, () => {
    });

    describe('watching enabled', () => {
-      const mockWatcher = newFSWatcherMock();
      beforeEach(async () => {
        configMock.load.mockResolvedValue(systemConfigStub.libraryWatchEnabled);
        libraryMock.getAll.mockResolvedValue([]);
        await sut.init();
-
-        storageMock.watch.mockReturnValue(mockWatcher);
      });

      it('should watch library', async () => {
        libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
        libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);

-        const mockWatcher = newFSWatcherMock();
-
-        let isReady = false;
-
-        mockWatcher.on.mockImplementation((event, callback) => {
-          if (event === 'ready') {
-            isReady = true;
-            callback();
-          }
-        });
-
-        storageMock.watch.mockReturnValue(mockWatcher);
-
        await sut.watchAll();

        expect(storageMock.watch).toHaveBeenCalledWith(
          libraryStub.externalLibraryWithImportPaths1.importPaths,
          expect.anything(),
+          expect.anything(),
        );
-
-        expect(isReady).toBe(true);
      });

      it('should watch and unwatch library', async () => {
        libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);
        libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
-
-        mockWatcher.on.mockImplementation((event, callback) => {
-          if (event === 'ready') {
-            callback();
-          }
-        });
+        const mockClose = jest.fn();
+        storageMock.watch.mockImplementation(makeMockWatcher({ close: mockClose }));

        await sut.watchAll();
        await sut.unwatch(libraryStub.externalLibraryWithImportPaths1.id);

-        expect(mockWatcher.close).toHaveBeenCalled();
+        expect(mockClose).toHaveBeenCalled();
      });

      it('should not watch library without import paths', async () => {
@@ -1277,14 +1212,7 @@ describe(LibraryService.name, () => {
      it('should handle a new file event', async () => {
        libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
        libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);
-
-        mockWatcher.on.mockImplementation((event, callback) => {
-          if (event === 'ready') {
-            callback();
-          } else if (event === 'add') {
-            callback('/foo/photo.jpg');
-          }
-        });
+        storageMock.watch.mockImplementation(makeMockWatcher({ items: [{ event: 'add', value: '/foo/photo.jpg' }] }));

        await sut.watchAll();

@@ -1304,14 +1232,9 @@ describe(LibraryService.name, () => {
      it('should handle a file change event', async () => {
        libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
        libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);
-
-        mockWatcher.on.mockImplementation((event, callback) => {
-          if (event === 'ready') {
-            callback();
-          } else if (event === 'change') {
-            callback('/foo/photo.jpg');
-          }
-        });
+        storageMock.watch.mockImplementation(
+          makeMockWatcher({ items: [{ event: 'change', value: '/foo/photo.jpg' }] }),
+        );

        await sut.watchAll();

@@ -1331,16 +1254,10 @@ describe(LibraryService.name, () => {
      it('should handle a file unlink event', async () => {
        libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
        libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);
-
        assetMock.getByLibraryIdAndOriginalPath.mockResolvedValue(assetStub.external);
-
-        mockWatcher.on.mockImplementation((event, callback) => {
-          if (event === 'ready') {
-            callback();
-          } else if (event === 'unlink') {
-            callback('/foo/photo.jpg');
-          }
-        });
+        storageMock.watch.mockImplementation(
+          makeMockWatcher({ items: [{ event: 'unlink', value: '/foo/photo.jpg' }] }),
+        );

        await sut.watchAll();

@@ -1351,34 +1268,19 @@ describe(LibraryService.name, () => {
        libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
        assetMock.getByLibraryIdAndOriginalPath.mockResolvedValue(assetStub.external);
        libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);
-
-        let didError = false;
-
-        mockWatcher.on.mockImplementation((event, callback) => {
-          if (event === 'ready') {
-            callback();
-          } else if (event === 'error') {
-            didError = true;
-            callback('Error!');
-          }
-        });
+        storageMock.watch.mockImplementation(
+          makeMockWatcher({
+            items: [{ event: 'error', value: 'Error!' }],
+          }),
+        );

        await sut.watchAll();
-
-        expect(didError).toBe(true);
      });

      it('should ignore unknown extensions', async () => {
        libraryMock.get.mockResolvedValue(libraryStub.externalLibraryWithImportPaths1);
        libraryMock.getAll.mockResolvedValue([libraryStub.externalLibraryWithImportPaths1]);
-
-        mockWatcher.on.mockImplementation((event, callback) => {
-          if (event === 'ready') {
-            callback();
-          } else if (event === 'add') {
-            callback('/foo/photo.txt');
-          }
-        });
+        storageMock.watch.mockImplementation(makeMockWatcher({ items: [{ event: 'add', value: '/foo/photo.jpg' }] }));

        await sut.watchAll();

@@ -1388,14 +1290,7 @@ describe(LibraryService.name, () => {
      it('should ignore excluded paths', async () => {
        libraryMock.get.mockResolvedValue(libraryStub.patternPath);
        libraryMock.getAll.mockResolvedValue([libraryStub.patternPath]);
-
-        mockWatcher.on.mockImplementation((event, callback) => {
-          if (event === 'ready') {
-            callback();
-          } else if (event === 'add') {
-            callback('/dir1/photo.txt');
-          }
-        });
+        storageMock.watch.mockImplementation(makeMockWatcher({ items: [{ event: 'add', value: '/dir1/photo.txt' }] }));

        await sut.watchAll();

@@ -1405,14 +1300,7 @@ describe(LibraryService.name, () => {
      it('should ignore excluded paths without case sensitivity', async () => {
        libraryMock.get.mockResolvedValue(libraryStub.patternPath);
        libraryMock.getAll.mockResolvedValue([libraryStub.patternPath]);
-
-        mockWatcher.on.mockImplementation((event, callback) => {
-          if (event === 'ready') {
-            callback();
-          } else if (event === 'add') {
-            callback('/DIR1/photo.txt');
-          }
-        });
+        storageMock.watch.mockImplementation(makeMockWatcher({ items: [{ event: 'add', value: '/DIR1/photo.txt' }] }));

        await sut.watchAll();

@@ -1445,20 +1333,13 @@ describe(LibraryService.name, () => {
        }
      });

-      const mockWatcher = newFSWatcherMock();
-
-      mockWatcher.on.mockImplementation((event, callback) => {
-        if (event === 'ready') {
-          callback();
-        }
-      });
-
-      storageMock.watch.mockReturnValue(mockWatcher);
+      const mockClose = jest.fn();
+      storageMock.watch.mockImplementation(makeMockWatcher({ close: mockClose }));

      await sut.init();
      await sut.unwatchAll();

-      expect(mockWatcher.close).toHaveBeenCalledTimes(2);
+      expect(mockClose).toHaveBeenCalledTimes(2);
    });
  });

--- a/server/src/domain/library/library.service.ts
+++ b/server/src/domain/library/library.service.ts
@@ -38,10 +38,8 @@ export class LibraryService extends EventEmitter {
  readonly logger = new ImmichLogger(LibraryService.name);
  private access: AccessCore;
  private configCore: SystemConfigCore;
-
  private watchLibraries = false;
-
-  private watchers: Record<string, () => Promise<void>> = {};
+  private watchers: Record<string, () => void> = {};

  constructor(
    @Inject(IAccessRepository) accessRepository: IAccessRepository,
@@ -116,63 +114,57 @@ export class LibraryService extends EventEmitter {

    this.logger.debug(`Settings for watcher: usePolling: ${usePolling}, interval: ${interval}`);

-    const watcher = this.storageRepository.watch(library.importPaths, {
-      usePolling,
-      interval,
-      binaryInterval: interval,
-      ignoreInitial: true,
-    });
+    let _resolve: () => void;
+    const ready$ = new Promise<void>((resolve) => (_resolve = resolve));

-    this.watchers[id] = async () => {
-      await watcher.close();
-    };
-
-    watcher.on('add', async (path) => {
-      this.logger.debug(`File add event received for ${path} in library ${library.id}}`);
-      if (matcher(path)) {
-        await this.scanAssets(library.id, [path], library.ownerId, false);
-      }
-      this.emit('add', path);
-    });
-
-    watcher.on('change', async (path) => {
-      this.logger.debug(`Detected file change for ${path} in library ${library.id}`);
-
-      if (matcher(path)) {
-        // Note: if the changed file was not previously imported, it will be imported now.
-        await this.scanAssets(library.id, [path], library.ownerId, false);
-      }
-      this.emit('change', path);
-    });
-
-    watcher.on('unlink', async (path) => {
-      this.logger.debug(`Detected deleted file at ${path} in library ${library.id}`);
-      const existingAssetEntity = await this.assetRepository.getByLibraryIdAndOriginalPath(library.id, path);
-
-      if (existingAssetEntity && matcher(path)) {
-        await this.assetRepository.save({ id: existingAssetEntity.id, isOffline: true });
-      }
-
-      this.emit('unlink', path);
-    });
-
-    watcher.on('error', async (error) => {
-      // TODO: should we log, or throw an exception?
-      this.logger.error(`Library watcher for library ${library.id} encountered error: ${error}`);
-    });
+    this.watchers[id] = this.storageRepository.watch(
+      library.importPaths,
+      {
+        usePolling,
+        interval,
+        binaryInterval: interval,
+        ignoreInitial: true,
+      },
+      {
+        onReady: () => _resolve(),
+        onAdd: async (path) => {
+          this.logger.debug(`File add event received for ${path} in library ${library.id}}`);
+          if (matcher(path)) {
+            await this.scanAssets(library.id, [path], library.ownerId, false);
+          }
+          this.emit('add', path);
+        },
+        onChange: async (path) => {
+          this.logger.debug(`Detected file change for ${path} in library ${library.id}`);
+          if (matcher(path)) {
+            // Note: if the changed file was not previously imported, it will be imported now.
+            await this.scanAssets(library.id, [path], library.ownerId, false);
+          }
+          this.emit('change', path);
+        },
+        onUnlink: async (path) => {
+          this.logger.debug(`Detected deleted file at ${path} in library ${library.id}`);
+          const asset = await this.assetRepository.getByLibraryIdAndOriginalPath(library.id, path);
+          if (asset && matcher(path)) {
+            await this.assetRepository.save({ id: asset.id, isOffline: true });
+          }
+          this.emit('unlink', path);
+        },
+        onError: (error) => {
+          // TODO: should we log, or throw an exception?
+          this.logger.error(`Library watcher for library ${library.id} encountered error: ${error}`);
+        },
+      },
+    );

    // Wait for the watcher to initialize before returning
-    await new Promise<void>((resolve) => {
-      watcher.on('ready', async () => {
-        resolve();
-      });
-    });
+    await ready$;

    return true;
  }

  async unwatch(id: string) {
-    if (this.watchers.hasOwnProperty(id)) {
+    if (this.watchers[id]) {
      await this.watchers[id]();
      delete this.watchers[id];
    }
--- a/server/src/domain/repositories/storage.repository.ts
+++ b/server/src/domain/repositories/storage.repository.ts
@@ -1,4 +1,4 @@
-import { FSWatcher, WatchOptions } from 'chokidar';
+import { WatchOptions } from 'chokidar';
 import { Stats } from 'node:fs';
 import { FileReadOptions } from 'node:fs/promises';
 import { Readable } from 'node:stream';
@@ -23,7 +23,13 @@ export interface DiskUsage {

 export const IStorageRepository = 'IStorageRepository';

-export interface ImmichWatcher extends FSWatcher {}
+export interface WatchEvents {
+  onReady(): void;
+  onAdd(path: string): void;
+  onChange(path: string): void;
+  onUnlink(path: string): void;
+  onError(error: Error): void;
+}

 export interface IStorageRepository {
  createZipStream(): ImmichZipStream;
@@ -41,6 +47,6 @@ export interface IStorageRepository {
  crawl(crawlOptions: CrawlOptionsDto): Promise<string[]>;
  copyFile(source: string, target: string): Promise<void>;
  rename(source: string, target: string): Promise<void>;
-  watch(paths: string[], options: WatchOptions): ImmichWatcher;
+  watch(paths: string[], options: WatchOptions, events: Partial<WatchEvents>): () => void;
  utimes(filepath: string, atime: Date, mtime: Date): Promise<void>;
 }
--- a/server/src/infra/repositories/filesystem.provider.ts
+++ b/server/src/infra/repositories/filesystem.provider.ts
@@ -2,10 +2,10 @@ import {
  CrawlOptionsDto,
  DiskUsage,
  ImmichReadStream,
-  ImmichWatcher,
  ImmichZipStream,
  IStorageRepository,
  mimeTypes,
+  WatchEvents,
 } from '@app/domain';
 import { ImmichLogger } from '@app/infra/logger';
 import archiver from 'archiver';
@@ -136,8 +136,15 @@ export class FilesystemProvider implements IStorageRepository {
    });
  }

-  watch(paths: string[], options: WatchOptions): ImmichWatcher {
-    return chokidar.watch(paths, options);
+  watch(paths: string[], options: WatchOptions, events: Partial<WatchEvents>) {
+    const watcher = chokidar.watch(paths, options);
+
+    watcher.on('ready', () => events.onReady?.());
+    watcher.on('add', (path) => events.onAdd?.(path));
+    watcher.on('change', (path) => events.onChange?.(path));
+    watcher.on('unlink', (path) => events.onUnlink?.(path));
+
+    return () => watcher.close();
  }

  readdir = readdir;