move to medium tests

server/src/repositories/event.repository.ts

@@ -95,10 +95,6 @@ type EventMap = {
 
   // websocket events
   WebsocketConnect: [{ userId: string }];
-
-  // library events
-  LibraryWatchEnabled: [{ id: string }];
-  LibraryWatchFired: [{ libraryId: string; event: 'add' | 'change' | 'unlink'; path: string; ignored: boolean }];
 };
 
 export type AppRestartEvent = {

server/src/repositories/websocket.repository.ts

@@ -35,9 +35,6 @@ export interface ClientEventMap {
   on_notification: [NotificationDto];
   on_session_delete: [string];
 
-  on_library_watch_enabled: [{ libraryId: string }];
-  on_library_watch_fired: [{ libraryId: string; event: 'add' | 'change' | 'unlink'; path: string; ignored: boolean }];
-
   AssetUploadReadyV1: [{ asset: SyncAssetV1; exif: SyncAssetExifV1 }];
   AppRestartV1: [AppRestartEvent];
   AssetEditReadyV1: [{ asset: SyncAssetV1; edit: SyncAssetEditV1[] }];

server/src/services/library.service.ts

@@ -107,13 +107,6 @@ export class LibraryService extends BaseService {
     const handler = async (event: string, path: string) => {
       const ignored = !isSupportedFile(path);
 
-      await this.eventRepository.emit('LibraryWatchFired', {
-        libraryId: library.id,
-        event: event as 'add' | 'change',
-        path,
-        ignored,
-      });
-
       if (ignored) {
         this.logger.verbose(`Ignoring file ${event} event for ${path} in library ${library.id}`);
         return;
@@ -127,13 +120,6 @@ export class LibraryService extends BaseService {
     };
 
     const deletionHandler = async (path: string) => {
-      await this.eventRepository.emit('LibraryWatchFired', {
-        libraryId: library.id,
-        event: 'unlink',
-        path,
-        ignored: false,
-      });
-
       this.logger.debug(`File unlink event received for ${path} in library ${library.id}}`);
       await this.jobRepository.queue({
         name: JobName.LibraryRemoveAsset,
@@ -172,8 +158,6 @@ export class LibraryService extends BaseService {
     // Wait for the watcher to initialize before returning
     await ready$;
 
-    await this.eventRepository.emit('LibraryWatchEnabled', { id });
-
     return true;
   }
 

server/src/services/notification.service.ts

@@ -114,16 +114,6 @@ export class NotificationService extends BaseService {
     this.websocketRepository.serverSend('ConfigUpdate', { oldConfig, newConfig });
   }
 
-  @OnEvent({ name: 'LibraryWatchEnabled' })
-  onLibraryWatchEnabled({ id }: ArgOf<'LibraryWatchEnabled'>) {
-    this.websocketRepository.clientBroadcast('on_library_watch_enabled', { libraryId: id });
-  }
-
-  @OnEvent({ name: 'LibraryWatchFired' })
-  onLibraryWatchFired(event: ArgOf<'LibraryWatchFired'>) {
-    this.websocketRepository.clientBroadcast('on_library_watch_fired', event);
-  }
-
   @OnEvent({ name: 'AppRestart' })
   onAppRestart(state: ArgOf<'AppRestart'>) {
     this.websocketRepository.clientBroadcast('AppRestartV1', {

server/test/medium.factory.ts

@@ -30,6 +30,7 @@ import { DatabaseRepository } from 'src/repositories/database.repository';
 import { EmailRepository } from 'src/repositories/email.repository';
 import { EventRepository } from 'src/repositories/event.repository';
 import { JobRepository } from 'src/repositories/job.repository';
+import { LibraryRepository } from 'src/repositories/library.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { MachineLearningRepository } from 'src/repositories/machine-learning.repository';
 import { MapRepository } from 'src/repositories/map.repository';
@@ -405,6 +406,7 @@ const newRealRepository = <T>(key: ClassConstructor<T>, db: Kysely<DB>): T => {
     case AssetRepository:
     case AssetEditRepository:
     case AssetJobRepository:
+    case LibraryRepository:
     case MemoryRepository:
     case NotificationRepository:
     case OcrRepository:
@@ -466,6 +468,7 @@ const newMockRepository = <T>(key: ClassConstructor<T>) => {
     case AlbumRepository:
     case AssetRepository:
     case AssetJobRepository:
+    case LibraryRepository:
     case ConfigRepository:
     case CryptoRepository:
     case MemoryRepository:

server/test/medium/specs/repositories/storage.repository.spec.ts

@@ -0,0 +1,135 @@
+import { Kysely } from 'kysely';
+import { mkdtemp, rm, unlink, writeFile } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { LoggingRepository } from 'src/repositories/logging.repository';
+import { StorageRepository } from 'src/repositories/storage.repository';
+import { DB } from 'src/schema';
+import { BaseService } from 'src/services/base.service';
+import { newMediumService } from 'test/medium.factory';
+import { getKyselyDB } from 'test/utils';
+
+let defaultDatabase: Kysely<DB>;
+
+const setup = (db?: Kysely<DB>) => {
+  const { ctx } = newMediumService(BaseService, {
+    database: db || defaultDatabase,
+    real: [],
+    mock: [LoggingRepository],
+  });
+
+  return { ctx, sut: ctx.get(StorageRepository) };
+};
+
+beforeAll(async () => {
+  defaultDatabase = await getKyselyDB();
+});
+
+const watchForEvent = (
+  sut: StorageRepository,
+  folder: string,
+  event: 'add' | 'change' | 'unlink',
+  action: () => Promise<void>,
+): Promise<string> => {
+  return new Promise<string>((resolve, reject) => {
+    const timeout = setTimeout(() => {
+      void close().finally(() => reject(new Error(`Timed out waiting for ${event} event`)));
+    }, 10_000);
+
+    const onResolve = (path: string) => {
+      clearTimeout(timeout);
+      void close().finally(() => resolve(path));
+    };
+
+    const onReject = (error: Error) => {
+      clearTimeout(timeout);
+      void close().finally(() => reject(error));
+    };
+
+    const close = sut.watch(
+      [folder],
+      {
+        ignoreInitial: true,
+        usePolling: true,
+        interval: 50,
+      },
+      {
+        onReady: () => {
+          void action().catch((error) => onReject(error as Error));
+        },
+        onAdd: (path) => {
+          if (event === 'add') {
+            onResolve(path);
+          }
+        },
+        onChange: (path) => {
+          if (event === 'change') {
+            onResolve(path);
+          }
+        },
+        onUnlink: (path) => {
+          if (event === 'unlink') {
+            onResolve(path);
+          }
+        },
+        onError: (error) => onReject(error),
+      },
+    );
+  });
+};
+
+describe(StorageRepository.name, () => {
+  describe('watch', () => {
+    it('should fire create (add) events', async () => {
+      const { sut } = setup();
+      const folder = await mkdtemp(join(tmpdir(), 'immich-storage-watch-add-'));
+      const file = join(folder, 'created.jpg');
+
+      try {
+        const changedPath = await watchForEvent(sut, folder, 'add', async () => {
+          await writeFile(file, 'created');
+        });
+
+        expect(changedPath).toBe(file);
+      } finally {
+        await rm(folder, { recursive: true, force: true });
+      }
+    });
+
+    it('should fire change events', async () => {
+      const { sut } = setup();
+      const folder = await mkdtemp(join(tmpdir(), 'immich-storage-watch-change-'));
+      const file = join(folder, 'changed.jpg');
+
+      await writeFile(file, 'before');
+
+      try {
+        const changedPath = await watchForEvent(sut, folder, 'change', async () => {
+          await writeFile(file, 'after');
+        });
+
+        expect(changedPath).toBe(file);
+      } finally {
+        await rm(folder, { recursive: true, force: true });
+      }
+    });
+
+    it('should fire unlink events', async () => {
+      const { sut } = setup();
+      const folder = await mkdtemp(join(tmpdir(), 'immich-storage-watch-unlink-'));
+      const file = join(folder, 'deleted.jpg');
+
+      await writeFile(file, 'content');
+
+      try {
+        const changedPath = await watchForEvent(sut, folder, 'unlink', async () => {
+          await unlink(file);
+        });
+
+        expect(changedPath).toBe(file);
+      } finally {
+        await rm(folder, { recursive: true, force: true });
+      }
+    });
+  });
+});

server/test/medium/specs/services/library.service.spec.ts

@@ -0,0 +1,169 @@
+import { ChokidarOptions } from 'chokidar';
+import { Kysely } from 'kysely';
+import { JobName } from 'src/enum';
+import { EventRepository } from 'src/repositories/event.repository';
+import { JobRepository } from 'src/repositories/job.repository';
+import { LibraryRepository } from 'src/repositories/library.repository';
+import { LoggingRepository } from 'src/repositories/logging.repository';
+import { StorageRepository, WatchEvents } from 'src/repositories/storage.repository';
+import { DB } from 'src/schema';
+import { LibraryService } from 'src/services/library.service';
+import { newMediumService } from 'test/medium.factory';
+import { getKyselyDB } from 'test/utils';
+
+let defaultDatabase: Kysely<DB>;
+
+const setup = (db?: Kysely<DB>) => {
+  return newMediumService(LibraryService, {
+    database: db || defaultDatabase,
+    real: [LibraryRepository],
+    mock: [EventRepository, JobRepository, LoggingRepository, StorageRepository],
+  });
+};
+
+beforeAll(async () => {
+  defaultDatabase = await getKyselyDB();
+});
+
+const closeWatcher = () => Promise.resolve();
+
+const setWatchMode = (sut: LibraryService) => {
+  const service = sut as unknown as { lock: boolean; watchLibraries: boolean };
+  service.lock = true;
+  service.watchLibraries = true;
+};
+
+const makeMockWatcher =
+  (paths: { add?: string[]; change?: string[]; unlink?: string[] }) =>
+  (_paths: string[], options: ChokidarOptions, events: Partial<WatchEvents>): (() => Promise<void>) => {
+    const ignored = options.ignored as ((path: string) => boolean) | undefined;
+    events.onReady?.();
+    for (const path of paths.add ?? []) {
+      if (!ignored?.(path)) {
+        events.onAdd?.(path);
+      }
+    }
+
+    for (const path of paths.change ?? []) {
+      if (!ignored?.(path)) {
+        events.onChange?.(path);
+      }
+    }
+
+    for (const path of paths.unlink ?? []) {
+      if (!ignored?.(path)) {
+        events.onUnlink?.(path);
+      }
+    }
+
+    return closeWatcher;
+  };
+
+describe(`${LibraryService.name} (watch, medium)`, () => {
+  it('should queue add and change events for supported files', async () => {
+    const { sut, ctx } = setup();
+    const storageRepo = ctx.getMock(StorageRepository);
+    const jobRepo = ctx.getMock(JobRepository);
+
+    jobRepo.queue.mockResolvedValue();
+    setWatchMode(sut);
+
+    const { user } = await ctx.newUser();
+    const library = await sut.create({
+      ownerId: user.id,
+      importPaths: ['/test-assets/temp/watcher-behavior'],
+      exclusionPatterns: ['**/@eaDir/**'],
+    });
+
+    storageRepo.watch.mockImplementation(
+      makeMockWatcher({
+        add: ['/test-assets/temp/watcher-behavior/add.png'],
+        change: ['/test-assets/temp/watcher-behavior/change.jpg'],
+      }),
+    );
+
+    await sut.watchAll();
+
+    expect(jobRepo.queue).toHaveBeenCalledWith({
+      name: JobName.LibrarySyncFiles,
+      data: { libraryId: library.id, paths: ['/test-assets/temp/watcher-behavior/add.png'] },
+    });
+    expect(jobRepo.queue).toHaveBeenCalledWith({
+      name: JobName.LibrarySyncFiles,
+      data: { libraryId: library.id, paths: ['/test-assets/temp/watcher-behavior/change.jpg'] },
+    });
+
+    await sut.onShutdown();
+  });
+
+  it('should queue unlink events for tracked files', async () => {
+    const { sut, ctx } = setup();
+    const storageRepo = ctx.getMock(StorageRepository);
+    const jobRepo = ctx.getMock(JobRepository);
+
+    jobRepo.queue.mockResolvedValue();
+    setWatchMode(sut);
+
+    const { user } = await ctx.newUser();
+    const library = await sut.create({
+      ownerId: user.id,
+      importPaths: ['/test-assets/temp/watcher-behavior'],
+      exclusionPatterns: ['**/@eaDir/**'],
+    });
+
+    storageRepo.watch.mockImplementation(
+      makeMockWatcher({
+        unlink: ['/test-assets/temp/watcher-behavior/delete.png'],
+      }),
+    );
+
+    await sut.watchAll();
+
+    expect(jobRepo.queue).toHaveBeenCalledWith({
+      name: JobName.LibraryRemoveAsset,
+      data: { libraryId: library.id, paths: ['/test-assets/temp/watcher-behavior/delete.png'] },
+    });
+
+    await sut.onShutdown();
+  });
+
+  it('should ignore add, change, and unlink events in excluded directories', async () => {
+    const { sut, ctx } = setup();
+    const storageRepo = ctx.getMock(StorageRepository);
+    const jobRepo = ctx.getMock(JobRepository);
+
+    jobRepo.queue.mockResolvedValue();
+    setWatchMode(sut);
+
+    await ctx.newUser().then(({ user }) =>
+      sut.create({
+        ownerId: user.id,
+        importPaths: ['/test-assets/temp/watcher-behavior'],
+        exclusionPatterns: ['**/@eaDir/**'],
+      }),
+    );
+
+    storageRepo.watch.mockImplementation(
+      makeMockWatcher({
+        add: ['/test-assets/temp/watcher-behavior/@eaDir/add.png'],
+        change: ['/test-assets/temp/watcher-behavior/@eaDir/change.png'],
+        unlink: ['/test-assets/temp/watcher-behavior/@eaDir/unlink.png'],
+      }),
+    );
+
+    await sut.watchAll();
+
+    expect(jobRepo.queue).not.toHaveBeenCalledWith(
+      expect.objectContaining({
+        name: JobName.LibrarySyncFiles,
+      }),
+    );
+    expect(jobRepo.queue).not.toHaveBeenCalledWith(
+      expect.objectContaining({
+        name: JobName.LibraryRemoveAsset,
+      }),
+    );
+
+    await sut.onShutdown();
+  });
+});