feat(rules): add watcher lists since media was added

apps/server/src/modules/api/media-server/emby/emby-adapter.service.spec.ts

@@ -96,6 +96,38 @@ describe('EmbyAdapterService', () => {
     setHttp();
   });
 
+  describe('getUsers', () => {
+    it('rethrows a failed lookup only when requested', async () => {
+      const error = new Error('boom');
+      http.get.mockRejectedValue(error);
+
+      await expect(service.getUsers()).resolves.toEqual([]);
+      await expect(service.getUsers(true)).rejects.toBe(error);
+    });
+
+    it('rejects a strict lookup when the API is not initialized', async () => {
+      (service as unknown as { http?: typeof http }).http = undefined;
+
+      await expect(service.getUsers(true)).rejects.toThrow(
+        'Emby API not initialized',
+      );
+    });
+
+    it.each([
+      ['missing Items', {}],
+      ['non-array Items', { Items: {} }],
+    ])(
+      'rejects malformed successful user responses with %s',
+      async (_description, data) => {
+        http.get.mockResolvedValue({ data });
+
+        await expect(service.getUsers(true)).rejects.toThrow();
+        await expect(service.getWatchHistory('item-1')).rejects.toThrow();
+        expect(embyCacheMocks.data.set).not.toHaveBeenCalled();
+      },
+    );
+  });
+
   describe('deleteFromDisk', () => {
     it.each(['', '   '])(
       'refuses a blank item id (%j) rather than calling /Items/',
@@ -381,7 +413,9 @@ describe('EmbyAdapterService', () => {
 
   describe('getChildrenMetadata caching (#3355)', () => {
     it('keys seasons and episodes of one parent separately', async () => {
-      http.get.mockResolvedValue({ data: { Items: [{ Id: 'child-1' }] } });
+      http.get.mockResolvedValue({
+        data: { Items: [{ Id: 'child-1' }], TotalRecordCount: 1 },
+      });
 
       await service.getChildrenMetadata('show-1', 'season');
       await service.getChildrenMetadata('show-1', 'episode');
@@ -415,6 +449,80 @@ describe('EmbyAdapterService', () => {
       ).resolves.toEqual([]);
       expect(embyCacheMocks.data.set).not.toHaveBeenCalled();
     });
+
+    it.each([
+      ['missing Items', {}],
+      ['a season without an id', { Items: [{ Type: 'Season' }] }],
+    ])(
+      'rejects strict season enumeration with %s',
+      async (_description, data) => {
+        http.get.mockResolvedValue({ data });
+
+        await expect(
+          service.getChildrenMetadata('show-1', 'season', true),
+        ).rejects.toThrow('Could not read the children of Emby item show-1');
+        expect(embyCacheMocks.data.set).not.toHaveBeenCalled();
+      },
+    );
+
+    it('pages past the batch limit instead of truncating children', async () => {
+      const page = (start: number, count: number) => ({
+        data: {
+          Items: Array.from({ length: count }, (_, index) => ({
+            Id: `episode-${start + index}`,
+            Type: 'Episode',
+          })),
+          TotalRecordCount: 501,
+        },
+      });
+      http.get
+        .mockResolvedValueOnce(page(0, 500))
+        .mockResolvedValueOnce(page(500, 1));
+
+      const children = await service.getChildrenMetadata('season-1', 'episode');
+
+      expect(children).toHaveLength(501);
+      expect(children[500].id).toBe('episode-500');
+      expect(http.get).toHaveBeenNthCalledWith(
+        2,
+        '/Items',
+        expect.objectContaining({
+          params: expect.objectContaining({ StartIndex: 500 }),
+        }),
+      );
+    });
+
+    it.each([
+      {
+        reason: 'the next page is empty',
+        secondPage: { Items: [], TotalRecordCount: 2 },
+      },
+      {
+        reason: 'the next page omits Items',
+        secondPage: { TotalRecordCount: 2 },
+      },
+      {
+        reason: 'the next page repeats an earlier item',
+        secondPage: {
+          Items: [{ Id: 'episode-1', Type: 'Episode' }],
+          TotalRecordCount: 2,
+        },
+      },
+    ])('rejects strict pagination when $reason', async ({ secondPage }) => {
+      http.get
+        .mockResolvedValueOnce({
+          data: {
+            Items: [{ Id: 'episode-1', Type: 'Episode' }],
+            TotalRecordCount: 2,
+          },
+        })
+        .mockResolvedValueOnce({ data: secondPage });
+
+      await expect(
+        service.getChildrenMetadata('season-1', 'episode', true),
+      ).rejects.toThrow('Could not read the children of Emby item season-1');
+      expect(embyCacheMocks.data.set).not.toHaveBeenCalled();
+    });
   });
 
   describe('getMetadata in-flight dedupe (#3356)', () => {
@@ -1394,6 +1502,61 @@ describe('EmbyAdapterService', () => {
     });
   });
 
+  describe('getDescendantEpisodeWatchHistory', () => {
+    const children = (parentId: string, ids: string[]) => ({
+      data: {
+        Items: ids.map((Id) => ({ Id, Type: 'Episode' })),
+        TotalRecordCount: ids.length,
+      },
+    });
+
+    it('keys every episode of a show by id with its per-user records', async () => {
+      http.get.mockImplementation(async (path: string, config?: any) => {
+        if (path === '/Shows/show-1/Seasons') {
+          return { data: { Items: [{ Id: 'season-1' }, { Id: 'season-2' }] } };
+        }
+        if (path === '/Items') {
+          return children(config.params.ParentId, [
+            `${config.params.ParentId}-ep`,
+          ]);
+        }
+        if (path === '/Users/Query') return { data: [{ Id: 'user-1' }] };
+        return {
+          data: {
+            UserData: {
+              Played: path.endsWith('season-1-ep'),
+              LastPlayedDate: '2024-06-02T00:00:00.000Z',
+            },
+          },
+        };
+      });
+
+      await expect(
+        service.getDescendantEpisodeWatchHistory('show-1', 'show'),
+      ).resolves.toEqual({
+        'season-1-ep': [
+          expect.objectContaining({
+            userId: 'user-1',
+            watchedAt: new Date('2024-06-02T00:00:00.000Z'),
+          }),
+        ],
+        'season-2-ep': [],
+      });
+    });
+
+    it('propagates a failed episode read instead of dropping the episode', async () => {
+      http.get.mockImplementation(async (path: string) => {
+        if (path === '/Items') return children('season-1', ['ep-1']);
+        if (path === '/Users/Query') return { data: [{ Id: 'user-1' }] };
+        throw createResponseError(502);
+      });
+
+      await expect(
+        service.getDescendantEpisodeWatchHistory('season-1', 'season'),
+      ).rejects.toThrow();
+    });
+  });
+
   describe('getLastPlayedAt', () => {
     const users = [
       { Id: 'user-1', Name: 'Alice' },

apps/server/src/modules/api/media-server/emby/emby-adapter.service.ts

@@ -207,8 +207,13 @@ export class EmbyAdapterService implements IMediaServerService {
     }
   }
 
-  async getUsers(): Promise<MediaUser[]> {
-    if (!this.http) return [];
+  async getUsers(throwOnError = false): Promise<MediaUser[]> {
+    if (!this.http) {
+      if (throwOnError) {
+        throw new Error('Emby API not initialized');
+      }
+      return [];
+    }
     try {
       const cached = this.cache.data.get<EmbyUserDto[]>(EMBY_CACHE_KEYS.USERS);
       const users = cached ? cached : await this.fetchUsersQuery(this.http);
@@ -220,6 +225,9 @@ export class EmbyAdapterService implements IMediaServerService {
       this.logger.debug(
         `Emby getUsers failed: ${formatConnectionFailureMessage(error, 'Connection failed')}`,
       );
+      if (throwOnError) {
+        throw error;
+      }
       return [];
     }
   }
@@ -590,29 +598,71 @@ export class EmbyAdapterService implements IMediaServerService {
             },
           },
         );
+        if (
+          !Array.isArray(data.Items) ||
+          data.Items.some((season) => !season.Id)
+        ) {
+          throw new Error('Emby returned invalid season items');
+        }
         return this.cacheChildren(
           cacheKey,
-          (data.Items ?? []).map(EmbyMapper.toMediaItem),
+          data.Items.map(EmbyMapper.toMediaItem),
         );
       }
 
-      const { data } = await this.http.get<EmbyItemsQueryResponse>('/Items', {
-        params: {
-          ParentId: parentId,
-          IncludeItemTypes: childType
-            ? EmbyMapper.toEmbyItemKind(childType)
-            : undefined,
-          // Skip virtual (unaired) episodes the same way the Jellyfin adapter does.
-          ExcludeLocationTypes: childType === 'episode' ? 'Virtual' : undefined,
-          Fields: 'ProviderIds,DateCreated,Overview,Tags',
-          EnableUserData: true,
-          Limit: EMBY_BATCH_SIZE.MAX_PAGE_SIZE,
-        },
-      });
-      return this.cacheChildren(
-        cacheKey,
-        (data.Items ?? []).map(EmbyMapper.toMediaItem),
-      );
+      const paginated = childType === 'episode';
+      const children: MediaItem[] = [];
+      const seenIds = new Set<string>();
+      let offset = 0;
+      let hasMore = true;
+
+      while (hasMore) {
+        const { data } = await this.http.get<EmbyItemsQueryResponse>('/Items', {
+          params: {
+            ParentId: parentId,
+            IncludeItemTypes: childType
+              ? EmbyMapper.toEmbyItemKind(childType)
+              : undefined,
+            // Skip virtual (unaired) episodes the same way the Jellyfin adapter does.
+            ExcludeLocationTypes:
+              childType === 'episode' ? 'Virtual' : undefined,
+            Fields: 'ProviderIds,DateCreated,Overview,Tags',
+            EnableUserData: true,
+            Limit: EMBY_BATCH_SIZE.MAX_PAGE_SIZE,
+            StartIndex: paginated ? offset : undefined,
+            EnableTotalRecordCount: paginated ? true : undefined,
+          },
+        });
+
+        if (paginated && !Array.isArray(data.Items)) {
+          throw new Error('Emby returned children without an Items list');
+        }
+        const items = data.Items ?? [];
+
+        if (paginated) {
+          if (
+            !Number.isSafeInteger(data.TotalRecordCount) ||
+            data.TotalRecordCount! < 0
+          ) {
+            throw new Error('Emby returned an invalid child count');
+          }
+          for (const item of items) {
+            if (!item.Id || seenIds.has(item.Id)) {
+              throw new Error('Emby returned duplicate child items');
+            }
+            seenIds.add(item.Id);
+          }
+        }
+
+        children.push(...items.map(EmbyMapper.toMediaItem));
+        offset += items.length;
+        hasMore = paginated && offset < data.TotalRecordCount!;
+        if (hasMore && items.length === 0) {
+          throw new Error('Emby child pagination made no progress');
+        }
+      }
+
+      return this.cacheChildren(cacheKey, children);
     } catch (error) {
       if (throwOnError) {
         // Worded like the Plex adapter's: the raw client error reaches the user
@@ -703,6 +753,52 @@ export class EmbyAdapterService implements IMediaServerService {
    * same question from its prefetched snapshot instead, because Emby omits the
    * watch dates a bulk sweep would need (see getWatchHistory).
    */
+  /**
+   * Watch records for every episode under `parentId`, keyed by episode id, the
+   * shape the Jellyfin adapter answers from its sweep. Emby has no dated bulk
+   * listing, so every episode costs one /Users/Query plus one
+   * /Users/{userId}/Items/{itemId} read per user, walked in batches.
+   * All-or-nothing: a failed read throws rather than answering with an
+   * episode missing from the map.
+   */
+  async getDescendantEpisodeWatchHistory(
+    parentId: string,
+    parentType: 'show' | 'season',
+  ): Promise<Record<string, WatchRecord[]>> {
+    const seasons =
+      parentType === 'season'
+        ? [{ id: parentId }]
+        : await this.getChildrenMetadata(parentId, 'season', true);
+    const episodeIds: string[] = [];
+    for (const season of seasons) {
+      const episodes = await this.getChildrenMetadata(
+        season.id,
+        'episode',
+        true,
+      );
+      episodeIds.push(...episodes.map((episode) => episode.id));
+    }
+
+    const watchHistory: Record<string, WatchRecord[]> = {};
+    for (
+      let i = 0;
+      i < episodeIds.length;
+      i += EMBY_BATCH_SIZE.EPISODE_WATCH_HISTORY
+    ) {
+      const batch = episodeIds.slice(
+        i,
+        i + EMBY_BATCH_SIZE.EPISODE_WATCH_HISTORY,
+      );
+      const records = await Promise.all(
+        batch.map((episodeId) => this.getWatchHistory(episodeId)),
+      );
+      batch.forEach((episodeId, index) => {
+        watchHistory[episodeId] = records[index];
+      });
+    }
+    return watchHistory;
+  }
+
   async getDescendantEpisodeWatchers(parentId: string): Promise<string[]> {
     if (!this.http) return [];
 
@@ -1877,7 +1973,9 @@ export class EmbyAdapterService implements IMediaServerService {
   private normalizeUsersResponse(
     data: EmbyUserDto[] | EmbyItemsQueryResponse<EmbyUserDto>,
   ): EmbyUserDto[] {
-    return Array.isArray(data) ? data : (data.Items ?? []);
+    if (Array.isArray(data)) return data;
+    if (Array.isArray(data.Items)) return data.Items;
+    throw new Error('Emby returned users without an Items list');
   }
 
   private buildAuthHeader(): string {

apps/server/src/modules/api/media-server/emby/emby.constants.ts

@@ -15,6 +15,7 @@ export const EMBY_CACHE_TTL = {
 
 export const EMBY_BATCH_SIZE = {
   USER_WATCH_HISTORY: 5,
+  EPISODE_WATCH_HISTORY: 5,
   COLLECTION_MUTATION: 8,
   DEFAULT_PAGE_SIZE: 100,
   MAX_PAGE_SIZE: 500,

apps/server/src/modules/api/media-server/emby/emby.mapper.spec.ts

@@ -248,6 +248,15 @@ describe('EmbyMapper', () => {
       expect(result.lastViewedAt).toEqual(new Date('2021-01-03T00:00:00.000Z'));
     });
 
+    it.each([undefined, 'not-a-date'])(
+      'keeps a missing or malformed DateCreated invalid',
+      (DateCreated) => {
+        const result = EmbyMapper.toMediaItem({ ...baseItem, DateCreated });
+
+        expect(Number.isNaN(result.addedAt.getTime())).toBe(true);
+      },
+    );
+
     it('converts RunTimeTicks (100-ns) to milliseconds', () => {
       const result = EmbyMapper.toMediaItem(baseItem);