diff --git a/server/api/plextv.ts b/server/api/plextv.ts index b8f3c9a47b..c417e772b5 100644 --- a/server/api/plextv.ts +++ b/server/api/plextv.ts @@ -6,6 +6,15 @@ import { randomUUID } from 'node:crypto'; import xml2js from 'xml2js'; import ExternalAPI from './externalapi'; +// Plex returns seconds, so fall back to a millisecond value if that ever changes. +const normalizeWatchlistedAt = (value: number | undefined): number | null => { + if (value === undefined || value <= 0) { + return null; + } + + return value > 1e11 ? Math.floor(value / 1000) : value; +}; + interface PlexAccountResponse { user: PlexUser; } @@ -122,6 +131,14 @@ interface MetadataResponse { }; } +interface UserStateResponse { + MediaContainer: { + UserState?: { + watchlistedAt?: number; + }[]; + }; +} + export interface PlexWatchlistItem { ratingKey: string; tmdbId: number; @@ -392,6 +409,43 @@ class PlexTvAPI extends ExternalAPI { } } + // Deliberately not this.get(), as ExternalAPI's cache key omits the auth token + // and would serve one user's watchlist state to another. + public async getWatchlistedAt(ratingKey: string): Promise { + const watchlistCache = cacheManager.getCache('plexwatchlist'); + const cacheKey = `userstate:${this.authToken}:${ratingKey}`; + const cached = watchlistCache.data.get(cacheKey); + + if (cached !== undefined) { + return cached; + } + + try { + const response = await this.axios.get( + `/library/metadata/${ratingKey}/userState`, + { + baseURL: 'https://discover.provider.plex.tv', + } + ); + + const watchlistedAt = normalizeWatchlistedAt( + response.data.MediaContainer.UserState?.[0]?.watchlistedAt + ); + + watchlistCache.data.set(cacheKey, watchlistedAt); + + return watchlistedAt; + } catch (e) { + logger.warn('Failed to retrieve watchlist state for item', { + label: 'Plex.TV Metadata API', + ratingKey, + errorMessage: e.message, + }); + + return null; + } + } + public async pingToken() { try { const response = await this.axios.get('/api/v2/ping', { diff --git a/server/datasource.ts b/server/datasource.ts index a87920f839..85175b2b3a 100644 --- a/server/datasource.ts +++ b/server/datasource.ts @@ -12,6 +12,8 @@ import { User } from '@server/entity/User'; import { UserPushSubscription } from '@server/entity/UserPushSubscription'; import { UserSettings } from '@server/entity/UserSettings'; import { Watchlist } from '@server/entity/Watchlist'; +import { WatchlistSyncEntry } from '@server/entity/WatchlistSyncEntry'; +import { WatchlistSyncEntrySeason } from '@server/entity/WatchlistSyncEntrySeason'; import { IssueCommentSubscriber } from '@server/subscriber/IssueCommentSubscriber'; import { IssueSubscriber } from '@server/subscriber/IssueSubscriber'; import { MediaRequestSubscriber } from '@server/subscriber/MediaRequestSubscriber'; @@ -39,6 +41,8 @@ const entities = [ UserPushSubscription, UserSettings, Watchlist, + WatchlistSyncEntry, + WatchlistSyncEntrySeason, ]; const subscribers = [ diff --git a/server/entity/WatchlistSyncEntry.ts b/server/entity/WatchlistSyncEntry.ts new file mode 100644 index 0000000000..8096fb17d6 --- /dev/null +++ b/server/entity/WatchlistSyncEntry.ts @@ -0,0 +1,58 @@ +import type { MediaType } from '@server/constants/media'; +import { User } from '@server/entity/User'; +import { WatchlistSyncEntrySeason } from '@server/entity/WatchlistSyncEntrySeason'; +import { DbAwareColumn, resolveDbType } from '@server/utils/DbColumnHelper'; +import { + Column, + Entity, + Index, + ManyToOne, + OneToMany, + PrimaryGeneratedColumn, + Unique, + UpdateDateColumn, +} from 'typeorm'; + +@Entity() +@Unique('UNIQUE_WATCHLIST_SYNC_ENTRY', ['requestedBy', 'tmdbId', 'mediaType']) +export class WatchlistSyncEntry { + @PrimaryGeneratedColumn() + public id: number; + + @Column() + @Index() + public tmdbId: number; + + @Column({ type: 'varchar' }) + public mediaType: MediaType; + + // Unix seconds, verbatim from Plex. + @Column({ type: 'int' }) + public watchlistedAt: number; + + @ManyToOne(() => User, { + onDelete: 'CASCADE', + nullable: false, + }) + @Index() + public requestedBy: User; + + @OneToMany(() => WatchlistSyncEntrySeason, (season) => season.entry, { + cascade: true, + eager: true, + }) + public seasons: WatchlistSyncEntrySeason[]; + + @DbAwareColumn({ type: 'datetime', default: () => 'CURRENT_TIMESTAMP' }) + public createdAt: Date; + + @UpdateDateColumn({ + type: resolveDbType('datetime'), + default: () => 'CURRENT_TIMESTAMP', + }) + public updatedAt: Date; + + constructor(init?: Partial) { + Object.assign(this, init); + } +} diff --git a/server/entity/WatchlistSyncEntrySeason.ts b/server/entity/WatchlistSyncEntrySeason.ts new file mode 100644 index 0000000000..e6ce59b8d4 --- /dev/null +++ b/server/entity/WatchlistSyncEntrySeason.ts @@ -0,0 +1,31 @@ +import { WatchlistSyncEntry } from '@server/entity/WatchlistSyncEntry'; +import { + Column, + Entity, + Index, + ManyToOne, + PrimaryGeneratedColumn, + Unique, +} from 'typeorm'; + +@Entity() +@Unique('UNIQUE_WATCHLIST_SYNC_ENTRY_SEASON', ['entry', 'seasonNumber']) +export class WatchlistSyncEntrySeason { + @PrimaryGeneratedColumn() + public id: number; + + @Column() + public seasonNumber: number; + + @ManyToOne(() => WatchlistSyncEntry, (entry) => entry.seasons, { + onDelete: 'CASCADE', + nullable: false, + orphanedRowAction: 'delete', + }) + @Index() + public entry: WatchlistSyncEntry; + + constructor(init?: Partial) { + Object.assign(this, init); + } +} diff --git a/server/job/schedule.ts b/server/job/schedule.ts index a9afd2f4d6..516bb85068 100644 --- a/server/job/schedule.ts +++ b/server/job/schedule.ts @@ -104,6 +104,7 @@ export const startJobs = (): void => { }); }); }), + running: () => watchlistSync.running, }); } else if ( mediaServerType === MediaServerType.JELLYFIN || diff --git a/server/lib/watchlistsync.test.ts b/server/lib/watchlistsync.test.ts index 58eceac0f2..ed4805645f 100644 --- a/server/lib/watchlistsync.test.ts +++ b/server/lib/watchlistsync.test.ts @@ -1,5 +1,6 @@ import type { PlexWatchlistItem } from '@server/api/plextv'; import PlexTvAPI from '@server/api/plextv'; +import TheMovieDb from '@server/api/themoviedb'; import { MediaRequestStatus, MediaStatus, @@ -7,15 +8,28 @@ import { } from '@server/constants/media'; import { getRepository } from '@server/datasource'; import Media from '@server/entity/Media'; -import { MediaRequest } from '@server/entity/MediaRequest'; +import { + DuplicateMediaRequestError, + MediaRequest, + NoSeasonsAvailableError, + QuotaRestrictedError, +} from '@server/entity/MediaRequest'; +import SeasonRequest from '@server/entity/SeasonRequest'; import { User } from '@server/entity/User'; import { UserSettings } from '@server/entity/UserSettings'; +import { WatchlistSyncEntry } from '@server/entity/WatchlistSyncEntry'; +import { WatchlistSyncEntrySeason } from '@server/entity/WatchlistSyncEntrySeason'; import { Permission } from '@server/lib/permissions'; import { setupTestDb } from '@server/test/db'; import assert from 'node:assert/strict'; import { beforeEach, describe, it } from 'node:test'; +const WATCHLISTED_AT = 1700000000; + let watchlistItems: PlexWatchlistItem[] = []; +let watchlistedAt = new Map(); +let userStateCalls: string[] = []; +let tmdbSeasons: number[] = []; Object.defineProperty(PlexTvAPI.prototype, 'getWatchlist', { get() { @@ -30,11 +44,51 @@ Object.defineProperty(PlexTvAPI.prototype, 'getWatchlist', { configurable: true, }); -let requestCalls: { mediaId: number; mediaType: MediaType }[] = []; +Object.defineProperty(PlexTvAPI.prototype, 'getWatchlistedAt', { + get() { + return async (ratingKey: string) => { + userStateCalls.push(ratingKey); + return watchlistedAt.has(ratingKey) + ? watchlistedAt.get(ratingKey) + : WATCHLISTED_AT; + }; + }, + set() {}, + configurable: true, +}); + +Object.defineProperty(TheMovieDb.prototype, 'getTvShow', { + get() { + return async ({ tvId }: { tvId: number }) => ({ + id: tvId, + seasons: tmdbSeasons.map((season_number) => ({ season_number })), + }); + }, + set() {}, + configurable: true, +}); + +type RequestCall = { + mediaId: number; + mediaType: MediaType; + seasons?: number[] | 'all'; +}; + +let requestCalls: RequestCall[] = []; +let requestError: Error | null = null; Object.defineProperty(MediaRequest, 'request', { - value: async (body: { mediaId: number; mediaType: MediaType }) => { - requestCalls.push({ mediaId: body.mediaId, mediaType: body.mediaType }); + value: async (body: RequestCall) => { + requestCalls.push({ + mediaId: body.mediaId, + mediaType: body.mediaType, + seasons: body.seasons, + }); + + if (requestError) { + throw requestError; + } + return {} as MediaRequest; }, writable: true, @@ -45,33 +99,31 @@ import watchlistSync from '@server/lib/watchlistsync'; setupTestDb(); -async function configureSyncUser(): Promise { +async function configureSyncUser(userId = 1): Promise { const userRepository = getRepository(User); - const admin = await userRepository.findOneOrFail({ where: { id: 1 } }); + const user = await userRepository.findOneOrFail({ where: { id: userId } }); - admin.plexToken = 'test-plex-token'; - admin.permissions = Permission.AUTO_REQUEST; - await userRepository.save(admin); + user.plexToken = 'test-plex-token'; + user.permissions = Permission.AUTO_REQUEST; + await userRepository.save(user); - const userSettingsRepository = getRepository(UserSettings); - await userSettingsRepository.save( + await getRepository(UserSettings).save( new UserSettings({ - user: admin, + user, watchlistSyncMovies: true, watchlistSyncTv: true, }) ); - return admin; + return user; } async function seedMedia( tmdbId: number, mediaType: MediaType, status: MediaStatus -): Promise { - const mediaRepository = getRepository(Media); - await mediaRepository.save( +): Promise { + return getRepository(Media).save( new Media({ tmdbId, mediaType, @@ -81,6 +133,71 @@ async function seedMedia( ); } +async function seedEntry( + user: User, + tmdbId: number, + mediaType: MediaType, + stamp: number, + seasons: number[] = [] +): Promise { + await getRepository(WatchlistSyncEntry).save( + new WatchlistSyncEntry({ + requestedBy: user, + tmdbId, + mediaType, + watchlistedAt: stamp, + seasons: seasons.map( + (seasonNumber) => new WatchlistSyncEntrySeason({ seasonNumber }) + ), + }) + ); +} + +async function seedRequest({ + user, + tmdbId, + mediaType, + createdAt, + seasons = [], + isAutoRequest = true, +}: { + user: User; + tmdbId: number; + mediaType: MediaType; + createdAt: number; + seasons?: number[]; + isAutoRequest?: boolean; +}): Promise { + const media = + (await getRepository(Media).findOne({ where: { tmdbId, mediaType } })) ?? + (await seedMedia(tmdbId, mediaType, MediaStatus.DELETED)); + + await getRepository(MediaRequest).save( + new MediaRequest({ + type: mediaType, + status: MediaRequestStatus.COMPLETED, + media, + requestedBy: user, + is4k: false, + isAutoRequest, + createdAt: new Date(createdAt * 1000), + seasons: seasons.map( + (seasonNumber) => new SeasonRequest({ seasonNumber }) + ), + }) + ); +} + +function getEntry( + tmdbId: number, + mediaType: MediaType, + userId = 1 +): Promise { + return getRepository(WatchlistSyncEntry).findOne({ + where: { requestedBy: { id: userId }, tmdbId, mediaType }, + }); +} + function movieItem(tmdbId: number, title: string): PlexWatchlistItem { return { ratingKey: `rk-${tmdbId}`, tmdbId, title, type: 'movie' }; } @@ -95,12 +212,20 @@ function showItem(tmdbId: number, title: string): PlexWatchlistItem { }; } -describe('WatchlistSync re-request gating', () => { - beforeEach(() => { - requestCalls = []; - watchlistItems = []; - }); +function requestedIds(): string[] { + return requestCalls.map((c) => `${c.mediaType}:${c.mediaId}`); +} +beforeEach(() => { + requestCalls = []; + watchlistItems = []; + watchlistedAt = new Map(); + userStateCalls = []; + tmdbSeasons = [0, 1, 2, 3]; + requestError = null; +}); + +describe('WatchlistSync status gating', () => { it('re-requests DELETED watchlist items and skips non-requestable ones', async () => { await configureSyncUser(); @@ -108,7 +233,6 @@ describe('WatchlistSync re-request gating', () => { await seedMedia(101, MediaType.MOVIE, MediaStatus.UNKNOWN); await seedMedia(102, MediaType.MOVIE, MediaStatus.AVAILABLE); await seedMedia(103, MediaType.MOVIE, MediaStatus.BLOCKLISTED); - await seedMedia(200, MediaType.TV, MediaStatus.DELETED); await seedMedia(201, MediaType.TV, MediaStatus.AVAILABLE); @@ -123,77 +247,331 @@ describe('WatchlistSync re-request gating', () => { await watchlistSync.syncWatchlist(); - const requestedArray = requestCalls.map( - (c) => `${c.mediaType}:${c.mediaId}` - ); - const requested = new Set(requestedArray); + const requested = new Set(requestedIds()); assert.strictEqual( - requestedArray.length, + requestedIds().length, requested.size, 'Each item should be requested exactly once' ); + assert.ok(requested.has(`${MediaType.MOVIE}:100`)); + assert.ok(requested.has(`${MediaType.MOVIE}:101`)); + assert.ok(!requested.has(`${MediaType.MOVIE}:102`)); + assert.ok(!requested.has(`${MediaType.MOVIE}:103`)); + assert.ok(requested.has(`${MediaType.TV}:200`)); + assert.ok(!requested.has(`${MediaType.TV}:201`)); + }); - assert.ok( - requested.has(`${MediaType.MOVIE}:100`), - 'DELETED movie on the watchlist should be re-requested' - ); + it('does not look up watchlist state for items the status filter drops', async () => { + await configureSyncUser(); + + await seedMedia(102, MediaType.MOVIE, MediaStatus.AVAILABLE); + + watchlistItems = [movieItem(101, 'New Movie'), movieItem(102, 'Available')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(userStateCalls, ['rk-101']); + }); +}); + +describe('WatchlistSync movie entries', () => { + it('requests a new entry and records it', async () => { + await configureSyncUser(); + watchlistItems = [movieItem(100, 'New Movie')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestedIds(), [`${MediaType.MOVIE}:100`]); + + const entry = await getEntry(100, MediaType.MOVIE); + assert.strictEqual(entry?.watchlistedAt, WATCHLISTED_AT); + }); + + it('skips an entry that has already been acted on', async () => { + const user = await configureSyncUser(); + await seedEntry(user, 100, MediaType.MOVIE, WATCHLISTED_AT); + + watchlistItems = [movieItem(100, 'Tracked Movie')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestCalls, []); + }); + + it('requests again when the title is re-added', async () => { + const user = await configureSyncUser(); + await seedEntry(user, 100, MediaType.MOVIE, WATCHLISTED_AT - 5000); + + watchlistItems = [movieItem(100, 'Re-added Movie')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestedIds(), [`${MediaType.MOVIE}:100`]); + + const entry = await getEntry(100, MediaType.MOVIE); + assert.strictEqual(entry?.watchlistedAt, WATCHLISTED_AT); + }); + + it('ignores request history when the title was re-added', async () => { + const user = await configureSyncUser(); + await seedEntry(user, 100, MediaType.MOVIE, WATCHLISTED_AT - 5000); + await seedRequest({ + user, + tmdbId: 100, + mediaType: MediaType.MOVIE, + createdAt: WATCHLISTED_AT + 60, + }); + + watchlistItems = [movieItem(100, 'Re-added Movie')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestedIds(), [`${MediaType.MOVIE}:100`]); + + const entry = await getEntry(100, MediaType.MOVIE); + assert.strictEqual(entry?.watchlistedAt, WATCHLISTED_AT); + }); + + it('records the entry when a request already exists', async () => { + await configureSyncUser(); + requestError = new DuplicateMediaRequestError('duplicate'); + + watchlistItems = [movieItem(100, 'Duplicate Movie')]; + + await watchlistSync.syncWatchlist(); + + const entry = await getEntry(100, MediaType.MOVIE); + assert.strictEqual(entry?.watchlistedAt, WATCHLISTED_AT); + }); + + it('retries on the next sync when the quota is exceeded', async () => { + await configureSyncUser(); + requestError = new QuotaRestrictedError('quota'); + + watchlistItems = [movieItem(100, 'Quota Movie')]; + + await watchlistSync.syncWatchlist(); - assert.ok( - requested.has(`${MediaType.MOVIE}:101`), - 'UNKNOWN movie should be requested' + assert.strictEqual(await getEntry(100, MediaType.MOVIE), null); + + requestError = null; + await watchlistSync.syncWatchlist(); + + assert.strictEqual(requestCalls.length, 2); + }); + + it('skips items with no watchlisted date', async () => { + await configureSyncUser(); + watchlistedAt.set('rk-100', null); + + watchlistItems = [movieItem(100, 'Undated Movie')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestCalls, []); + assert.strictEqual(await getEntry(100, MediaType.MOVIE), null); + }); +}); + +describe('WatchlistSync show entries', () => { + it('requests every season except specials and records them', async () => { + await configureSyncUser(); + watchlistItems = [showItem(200, 'New Show')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestCalls[0]?.seasons, [1, 2, 3]); + + const entry = await getEntry(200, MediaType.TV); + assert.deepStrictEqual( + entry?.seasons.map((s) => s.seasonNumber).sort((a, b) => a - b), + [1, 2, 3] ); - assert.ok( - !requested.has(`${MediaType.MOVIE}:102`), - 'AVAILABLE movie should NOT be requested' + }); + + it('requests only seasons added since the entry was acted on', async () => { + const user = await configureSyncUser(); + await seedEntry(user, 200, MediaType.TV, WATCHLISTED_AT, [1, 2, 3]); + tmdbSeasons = [0, 1, 2, 3, 4]; + + watchlistItems = [showItem(200, 'Ongoing Show')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestCalls[0]?.seasons, [4]); + + const entry = await getEntry(200, MediaType.TV); + assert.deepStrictEqual( + entry?.seasons.map((s) => s.seasonNumber).sort((a, b) => a - b), + [1, 2, 3, 4] ); - assert.ok( - !requested.has(`${MediaType.MOVIE}:103`), - 'BLOCKLISTED movie should NOT be requested' + }); + + it('does not request when every season has been acted on', async () => { + const user = await configureSyncUser(); + await seedEntry(user, 200, MediaType.TV, WATCHLISTED_AT, [1, 2, 3]); + + watchlistItems = [showItem(200, 'Handled Show')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestCalls, []); + }); + + it('requests all seasons again when the show is re-added', async () => { + const user = await configureSyncUser(); + await seedEntry(user, 200, MediaType.TV, WATCHLISTED_AT - 5000, [1, 2, 3]); + + watchlistItems = [showItem(200, 'Re-added Show')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestCalls[0]?.seasons, [1, 2, 3]); + + const entry = await getEntry(200, MediaType.TV); + assert.strictEqual(entry?.watchlistedAt, WATCHLISTED_AT); + assert.deepStrictEqual( + entry?.seasons.map((s) => s.seasonNumber).sort((a, b) => a - b), + [1, 2, 3] ); + }); + + it('ignores request history when the show was re-added', async () => { + const user = await configureSyncUser(); + await seedEntry(user, 200, MediaType.TV, WATCHLISTED_AT - 5000, [1, 2, 3]); + await seedRequest({ + user, + tmdbId: 200, + mediaType: MediaType.TV, + createdAt: WATCHLISTED_AT + 60, + seasons: [1, 2, 3], + }); + + watchlistItems = [showItem(200, 'Re-added Show')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestCalls[0]?.seasons, [1, 2, 3]); - assert.ok( - requested.has(`${MediaType.TV}:200`), - 'DELETED show should be re-requested' + const entry = await getEntry(200, MediaType.TV); + assert.strictEqual(entry?.watchlistedAt, WATCHLISTED_AT); + }); + + it('drops recorded seasons that no longer exist on the show', async () => { + const user = await configureSyncUser(); + await seedEntry(user, 200, MediaType.TV, WATCHLISTED_AT - 5000, [1, 2, 3]); + tmdbSeasons = [0, 1]; + + watchlistItems = [showItem(200, 'Shrunk Show')]; + + await watchlistSync.syncWatchlist(); + + const entry = await getEntry(200, MediaType.TV); + assert.deepStrictEqual( + entry?.seasons.map((s) => s.seasonNumber).sort((a, b) => a - b), + [1] ); - assert.ok( - !requested.has(`${MediaType.TV}:201`), - 'AVAILABLE show should NOT be requested' + }); + + it('records seasons that were already covered elsewhere', async () => { + await configureSyncUser(); + requestError = new NoSeasonsAvailableError('none'); + + watchlistItems = [showItem(200, 'Covered Show')]; + + await watchlistSync.syncWatchlist(); + + const entry = await getEntry(200, MediaType.TV); + assert.deepStrictEqual( + entry?.seasons.map((s) => s.seasonNumber).sort((a, b) => a - b), + [1, 2, 3] ); }); +}); - it('re-requests DELETED watchlist items even when a stale auto-request exists', async () => { +describe('WatchlistSync seeding from request history', () => { + it('records an entry instead of requesting when sync already handled it', async () => { const user = await configureSyncUser(); + await seedRequest({ + user, + tmdbId: 100, + mediaType: MediaType.MOVIE, + createdAt: WATCHLISTED_AT + 60, + }); - await seedMedia(100, MediaType.MOVIE, MediaStatus.DELETED); + watchlistItems = [movieItem(100, 'Previously Handled')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestCalls, []); + + const entry = await getEntry(100, MediaType.MOVIE); + assert.strictEqual(entry?.watchlistedAt, WATCHLISTED_AT); + }); - const media = await getRepository(Media).findOneOrFail({ - where: { tmdbId: 100, mediaType: MediaType.MOVIE }, + it('requests when the earlier request predates the watchlist add', async () => { + const user = await configureSyncUser(); + await seedRequest({ + user, + tmdbId: 100, + mediaType: MediaType.MOVIE, + createdAt: WATCHLISTED_AT - 5000, }); - await getRepository(MediaRequest).save( - new MediaRequest({ - type: MediaType.MOVIE, - status: MediaRequestStatus.COMPLETED, - media, - requestedBy: user, - is4k: false, - isAutoRequest: true, - }) - ); + watchlistItems = [movieItem(100, 'Re-added Since')]; - watchlistItems = [movieItem(100, 'Deleted Movie')]; + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestedIds(), [`${MediaType.MOVIE}:100`]); + }); + + it('ignores manual requests when seeding', async () => { + const user = await configureSyncUser(); + await seedRequest({ + user, + tmdbId: 100, + mediaType: MediaType.MOVIE, + createdAt: WATCHLISTED_AT + 60, + isAutoRequest: false, + }); + + watchlistItems = [movieItem(100, 'Manually Requested')]; await watchlistSync.syncWatchlist(); - const calls = requestCalls.filter( - (c) => c.mediaType === MediaType.MOVIE && c.mediaId === 100 - ); + assert.deepStrictEqual(requestedIds(), [`${MediaType.MOVIE}:100`]); + }); - assert.strictEqual( - calls.length, - 1, - 'DELETED movie should be re-requested even when a stale auto-request exists' - ); + it('seeds handled seasons from the earlier request', async () => { + const user = await configureSyncUser(); + await seedRequest({ + user, + tmdbId: 200, + mediaType: MediaType.TV, + createdAt: WATCHLISTED_AT + 60, + seasons: [1, 2], + }); + + watchlistItems = [showItem(200, 'Partly Handled Show')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestCalls[0]?.seasons, [3]); + }); +}); + +describe('WatchlistSync users', () => { + it('tracks the same title separately for each user', async () => { + const admin = await configureSyncUser(1); + await configureSyncUser(2); + await seedEntry(admin, 100, MediaType.MOVIE, WATCHLISTED_AT); + + watchlistItems = [movieItem(100, 'Shared Movie')]; + + await watchlistSync.syncWatchlist(); + + assert.deepStrictEqual(requestedIds(), [`${MediaType.MOVIE}:100`]); + assert.ok(await getEntry(100, MediaType.MOVIE, 2)); }); }); diff --git a/server/lib/watchlistsync.ts b/server/lib/watchlistsync.ts index cb8a5d35e2..06257160c7 100644 --- a/server/lib/watchlistsync.ts +++ b/server/lib/watchlistsync.ts @@ -1,4 +1,6 @@ +import type { PlexWatchlistItem } from '@server/api/plextv'; import PlexTvAPI from '@server/api/plextv'; +import TheMovieDb from '@server/api/themoviedb'; import { MediaStatus, MediaType } from '@server/constants/media'; import { getRepository } from '@server/datasource'; import Media from '@server/entity/Media'; @@ -11,23 +13,51 @@ import { RequestPermissionError, } from '@server/entity/MediaRequest'; import { User } from '@server/entity/User'; +import { WatchlistSyncEntry } from '@server/entity/WatchlistSyncEntry'; +import { WatchlistSyncEntrySeason } from '@server/entity/WatchlistSyncEntrySeason'; import logger from '@server/logger'; +import { In } from 'typeorm'; import { Permission } from './permissions'; +const CLOCK_SKEW_SECONDS = 300; + class WatchlistSync { + public running = false; + public async syncWatchlist() { - const userRepository = getRepository(User); - - // Get users who actually have plex tokens - const users = await userRepository - .createQueryBuilder('user') - .addSelect('user.plexToken') - .leftJoinAndSelect('user.settings', 'settings') - .where("user.plexToken != ''") - .getMany(); - - for (const user of users) { - await this.syncUserWatchlist(user); + if (this.running) { + logger.warn('Watchlist sync is already running, skipping this run', { + label: 'Watchlist Sync', + }); + return; + } + + this.running = true; + + try { + const userRepository = getRepository(User); + + // Get users who actually have plex tokens + const users = await userRepository + .createQueryBuilder('user') + .addSelect('user.plexToken') + .leftJoinAndSelect('user.settings', 'settings') + .where("user.plexToken != ''") + .getMany(); + + for (const user of users) { + try { + await this.syncUserWatchlist(user); + } catch (e) { + logger.error('Failed to sync watchlist for user', { + label: 'Watchlist Sync', + userId: user.id, + errorMessage: e instanceof Error ? e.message : String(e), + }); + } + } + } finally { + this.running = false; } } @@ -81,6 +111,7 @@ class WatchlistSync { ? await requestRepository .createQueryBuilder('request') .leftJoinAndSelect('request.media', 'media') + .leftJoinAndSelect('request.seasons', 'seasons') .where('request.requestedBy = :userId', { userId: user.id }) .andWhere('request.isAutoRequest = true') .andWhere('media.tmdbId IN (:...tmdbIds)', { @@ -116,7 +147,45 @@ class WatchlistSync { ); }); + if (!unavailableItems.length) { + return; + } + + const entries = await getRepository(WatchlistSyncEntry).find({ + where: { + requestedBy: { id: user.id }, + tmdbId: In(unavailableItems.map((i) => i.tmdbId)), + }, + }); + + const entriesByKey = new Map( + entries.map((e) => [`${e.mediaType}:${e.tmdbId}`, e]) + ); + + const autoRequestsByKey = new Map(); + for (const request of existingAutoRequests) { + if (!request.media) { + continue; + } + + const key = `${request.media.mediaType}:${request.media.tmdbId}`; + autoRequestsByKey.set(key, [ + ...(autoRequestsByKey.get(key) ?? []), + request, + ]); + } + + const tmdb = new TheMovieDb(); + for (const mediaItem of unavailableItems) { + const mediaType = + mediaItem.type === 'show' ? MediaType.TV : MediaType.MOVIE; + const key = `${mediaType}:${mediaItem.tmdbId}`; + + let entryStamp: number | null = null; + let handledSeasons: number[] = []; + let candidateSeasons: number[] = []; + try { if (mediaItem.type === 'show' && !mediaItem.tvdbId) { throw new Error('Missing TVDB ID from Plex Metadata'); @@ -141,12 +210,113 @@ class WatchlistSync { continue; } + const watchlistedAt = await plexTvApi.getWatchlistedAt( + mediaItem.ratingKey + ); + + if (!watchlistedAt) { + logger.warn('Skipping watchlist item without a watchlisted date', { + label: 'Watchlist Sync', + userId: user.id, + mediaTitle: mediaItem.title, + }); + continue; + } + + entryStamp = watchlistedAt; + + const entry = entriesByKey.get(key); + const rearmed = !!entry && entry.watchlistedAt !== watchlistedAt; + let handledPreviously = false; + + if (entry && rearmed) { + logger.info('Watchlist entry re-added, requesting again', { + label: 'Watchlist Sync', + userId: user.id, + mediaTitle: mediaItem.title, + previousWatchlistedAt: entry.watchlistedAt, + watchlistedAt, + }); + } + + if (entry && !rearmed) { + handledSeasons = entry.seasons.map((s) => s.seasonNumber); + } else if (!entry) { + // An auto-request made after the watchlist add means sync already acted on this entry. + const priorRequests = (autoRequestsByKey.get(key) ?? []).filter( + (r) => + Math.floor(r.createdAt.getTime() / 1000) >= + watchlistedAt - CLOCK_SKEW_SECONDS + ); + + handledPreviously = priorRequests.length > 0; + handledSeasons = this.mergeSeasons( + priorRequests.flatMap((r) => + (r.seasons ?? []).map((s) => s.seasonNumber) + ) + ); + } + + if (mediaType === MediaType.MOVIE) { + if (entry && !rearmed) { + continue; + } + + if (handledPreviously) { + await this.saveEntry(user, mediaItem, mediaType, watchlistedAt, []); + continue; + } + + await MediaRequest.request( + { + mediaId: mediaItem.tmdbId, + mediaType, + tvdbId: mediaItem.tvdbId, + is4k: false, + }, + user, + { isAutoRequest: true } + ); + + await this.saveEntry(user, mediaItem, mediaType, watchlistedAt, []); + + logger.info("Created media request from user's Plex Watchlist", { + label: 'Watchlist Sync', + userId: user.id, + mediaTitle: mediaItem.title, + }); + + continue; + } + + // Mirrors how 'all' expands in MediaRequest.request, which always drops season 0. + const tmdbShow = await tmdb.getTvShow({ tvId: mediaItem.tmdbId }); + candidateSeasons = tmdbShow.seasons + .filter((season) => season.season_number !== 0) + .map((season) => season.season_number); + + const requestedSeasons = candidateSeasons.filter( + (season) => !handledSeasons.includes(season) + ); + + if (!requestedSeasons.length) { + if (!entry || rearmed || handledPreviously) { + await this.saveEntry( + user, + mediaItem, + mediaType, + watchlistedAt, + this.mergeSeasons(handledSeasons, candidateSeasons) + ); + } + continue; + } + await MediaRequest.request( { mediaId: mediaItem.tmdbId, - mediaType: - mediaItem.type === 'show' ? MediaType.TV : MediaType.MOVIE, - seasons: mediaItem.type === 'show' ? 'all' : undefined, + mediaType, + seasons: requestedSeasons, tvdbId: mediaItem.tvdbId, is4k: false, }, @@ -154,6 +324,14 @@ class WatchlistSync { { isAutoRequest: true } ); + await this.saveEntry( + user, + mediaItem, + mediaType, + watchlistedAt, + this.mergeSeasons(handledSeasons, candidateSeasons) + ); + logger.info("Created media request from user's Plex Watchlist", { label: 'Watchlist Sync', userId: user.id, @@ -165,14 +343,31 @@ class WatchlistSync { } switch (e.constructor) { - // During watchlist sync, these errors aren't necessarily - // a problem with Seerr. Since we are auto syncing these constantly, it's - // possible they are unexpectedly at their quota limit, for example. So we'll - // instead log these as debug messages. - case RequestPermissionError: + // Nothing new was requested but the entry has still been acted on, so record it. case DuplicateMediaRequestError: - case QuotaRestrictedError: case NoSeasonsAvailableError: + if (entryStamp) { + await this.saveEntry( + user, + mediaItem, + mediaType, + entryStamp, + mediaType === MediaType.TV + ? this.mergeSeasons(handledSeasons, candidateSeasons) + : [] + ); + } + + logger.debug('Failed to create media request from watchlist', { + label: 'Watchlist Sync', + userId: user.id, + mediaTitle: mediaItem.title, + errorMessage: e.message, + }); + break; + // Leave the entry untracked so a later sync can retry + case RequestPermissionError: + case QuotaRestrictedError: logger.debug('Failed to create media request from watchlist', { label: 'Watchlist Sync', userId: user.id, @@ -194,6 +389,57 @@ class WatchlistSync { } } } + + private mergeSeasons(...seasons: number[][]): number[] { + return [...new Set(seasons.flat())]; + } + + private async saveEntry( + user: User, + mediaItem: PlexWatchlistItem, + mediaType: MediaType, + watchlistedAt: number, + seasons: number[] + ): Promise { + const entryRepository = getRepository(WatchlistSyncEntry); + + try { + const entry = + (await entryRepository.findOne({ + where: { + requestedBy: { id: user.id }, + tmdbId: mediaItem.tmdbId, + mediaType, + }, + })) ?? + new WatchlistSyncEntry({ + requestedBy: user, + tmdbId: mediaItem.tmdbId, + mediaType, + }); + + // Reuse the persisted rows so re-saving an entry does not collide with + // the unique constraint on its seasons. + const existingSeasons = entry.seasons ?? []; + + entry.watchlistedAt = watchlistedAt; + entry.seasons = seasons.map( + (seasonNumber) => + existingSeasons.find((s) => s.seasonNumber === seasonNumber) ?? + new WatchlistSyncEntrySeason({ seasonNumber }) + ); + + await entryRepository.save(entry); + } catch (e) { + // A concurrent run may have written the same entry first + logger.warn('Failed to record watchlist entry', { + label: 'Watchlist Sync', + userId: user.id, + mediaTitle: mediaItem.title, + errorMessage: e.message, + }); + } + } } const watchlistSync = new WatchlistSync(); diff --git a/server/migration/postgres/1786618023582-AddWatchlistSyncEntry.ts b/server/migration/postgres/1786618023582-AddWatchlistSyncEntry.ts new file mode 100644 index 0000000000..97716da38c --- /dev/null +++ b/server/migration/postgres/1786618023582-AddWatchlistSyncEntry.ts @@ -0,0 +1,49 @@ +import type { MigrationInterface, QueryRunner } from 'typeorm'; + +export class AddWatchlistSyncEntry1786618023582 implements MigrationInterface { + name = 'AddWatchlistSyncEntry1786618023582'; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query( + `CREATE TABLE "watchlist_sync_entry_season" ("id" SERIAL NOT NULL, "seasonNumber" integer NOT NULL, "entryId" integer NOT NULL, CONSTRAINT "UNIQUE_WATCHLIST_SYNC_ENTRY_SEASON" UNIQUE ("entryId", "seasonNumber"), CONSTRAINT "PK_2dc191c81c33714bf0c82c8adc2" PRIMARY KEY ("id"))` + ); + await queryRunner.query( + `CREATE INDEX "IDX_d9d29444df017aa0cbfa5a2244" ON "watchlist_sync_entry_season" ("entryId") ` + ); + await queryRunner.query( + `CREATE TABLE "watchlist_sync_entry" ("id" SERIAL NOT NULL, "tmdbId" integer NOT NULL, "mediaType" character varying NOT NULL, "watchlistedAt" integer NOT NULL, "createdAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), "updatedAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), "requestedById" integer NOT NULL, CONSTRAINT "UNIQUE_WATCHLIST_SYNC_ENTRY" UNIQUE ("requestedById", "tmdbId", "mediaType"), CONSTRAINT "PK_9f6976e1fc5d9d4266060c7d66c" PRIMARY KEY ("id"))` + ); + await queryRunner.query( + `CREATE INDEX "IDX_76036f4009dac0f28dd3913012" ON "watchlist_sync_entry" ("tmdbId") ` + ); + await queryRunner.query( + `CREATE INDEX "IDX_acad2425a838db492a20431e32" ON "watchlist_sync_entry" ("requestedById") ` + ); + await queryRunner.query( + `ALTER TABLE "watchlist_sync_entry_season" ADD CONSTRAINT "FK_d9d29444df017aa0cbfa5a2244d" FOREIGN KEY ("entryId") REFERENCES "watchlist_sync_entry"("id") ON DELETE CASCADE ON UPDATE NO ACTION` + ); + await queryRunner.query( + `ALTER TABLE "watchlist_sync_entry" ADD CONSTRAINT "FK_acad2425a838db492a20431e327" FOREIGN KEY ("requestedById") REFERENCES "user"("id") ON DELETE CASCADE ON UPDATE NO ACTION` + ); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query( + `ALTER TABLE "watchlist_sync_entry" DROP CONSTRAINT "FK_acad2425a838db492a20431e327"` + ); + await queryRunner.query( + `ALTER TABLE "watchlist_sync_entry_season" DROP CONSTRAINT "FK_d9d29444df017aa0cbfa5a2244d"` + ); + await queryRunner.query( + `DROP INDEX "public"."IDX_acad2425a838db492a20431e32"` + ); + await queryRunner.query( + `DROP INDEX "public"."IDX_76036f4009dac0f28dd3913012"` + ); + await queryRunner.query(`DROP TABLE "watchlist_sync_entry"`); + await queryRunner.query( + `DROP INDEX "public"."IDX_d9d29444df017aa0cbfa5a2244"` + ); + await queryRunner.query(`DROP TABLE "watchlist_sync_entry_season"`); + } +} diff --git a/server/migration/sqlite/1786620022403-AddWatchlistSyncEntry.ts b/server/migration/sqlite/1786620022403-AddWatchlistSyncEntry.ts new file mode 100644 index 0000000000..cf83e2d8b8 --- /dev/null +++ b/server/migration/sqlite/1786620022403-AddWatchlistSyncEntry.ts @@ -0,0 +1,97 @@ +import type { MigrationInterface, QueryRunner } from 'typeorm'; + +export class AddWatchlistSyncEntry1786620022403 implements MigrationInterface { + name = 'AddWatchlistSyncEntry1786620022403'; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query( + `CREATE TABLE "watchlist_sync_entry" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "tmdbId" integer NOT NULL, "mediaType" varchar NOT NULL, "watchlistedAt" integer NOT NULL, "createdAt" datetime NOT NULL DEFAULT (CURRENT_TIMESTAMP), "updatedAt" datetime NOT NULL DEFAULT (CURRENT_TIMESTAMP), "requestedById" integer NOT NULL, CONSTRAINT "UNIQUE_WATCHLIST_SYNC_ENTRY" UNIQUE ("requestedById", "tmdbId", "mediaType"))` + ); + await queryRunner.query( + `CREATE INDEX "IDX_76036f4009dac0f28dd3913012" ON "watchlist_sync_entry" ("tmdbId") ` + ); + await queryRunner.query( + `CREATE INDEX "IDX_acad2425a838db492a20431e32" ON "watchlist_sync_entry" ("requestedById") ` + ); + await queryRunner.query( + `CREATE TABLE "watchlist_sync_entry_season" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "seasonNumber" integer NOT NULL, "entryId" integer NOT NULL, CONSTRAINT "UNIQUE_WATCHLIST_SYNC_ENTRY_SEASON" UNIQUE ("entryId", "seasonNumber"))` + ); + await queryRunner.query( + `CREATE INDEX "IDX_d9d29444df017aa0cbfa5a2244" ON "watchlist_sync_entry_season" ("entryId") ` + ); + await queryRunner.query(`DROP INDEX "IDX_76036f4009dac0f28dd3913012"`); + await queryRunner.query(`DROP INDEX "IDX_acad2425a838db492a20431e32"`); + await queryRunner.query( + `CREATE TABLE "temporary_watchlist_sync_entry" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "tmdbId" integer NOT NULL, "mediaType" varchar NOT NULL, "watchlistedAt" integer NOT NULL, "createdAt" datetime NOT NULL DEFAULT (CURRENT_TIMESTAMP), "updatedAt" datetime NOT NULL DEFAULT (CURRENT_TIMESTAMP), "requestedById" integer NOT NULL, CONSTRAINT "UNIQUE_WATCHLIST_SYNC_ENTRY" UNIQUE ("requestedById", "tmdbId", "mediaType"), CONSTRAINT "FK_acad2425a838db492a20431e327" FOREIGN KEY ("requestedById") REFERENCES "user" ("id") ON DELETE CASCADE ON UPDATE NO ACTION)` + ); + await queryRunner.query( + `INSERT INTO "temporary_watchlist_sync_entry"("id", "tmdbId", "mediaType", "watchlistedAt", "createdAt", "updatedAt", "requestedById") SELECT "id", "tmdbId", "mediaType", "watchlistedAt", "createdAt", "updatedAt", "requestedById" FROM "watchlist_sync_entry"` + ); + await queryRunner.query(`DROP TABLE "watchlist_sync_entry"`); + await queryRunner.query( + `ALTER TABLE "temporary_watchlist_sync_entry" RENAME TO "watchlist_sync_entry"` + ); + await queryRunner.query( + `CREATE INDEX "IDX_76036f4009dac0f28dd3913012" ON "watchlist_sync_entry" ("tmdbId") ` + ); + await queryRunner.query( + `CREATE INDEX "IDX_acad2425a838db492a20431e32" ON "watchlist_sync_entry" ("requestedById") ` + ); + await queryRunner.query(`DROP INDEX "IDX_d9d29444df017aa0cbfa5a2244"`); + await queryRunner.query( + `CREATE TABLE "temporary_watchlist_sync_entry_season" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "seasonNumber" integer NOT NULL, "entryId" integer NOT NULL, CONSTRAINT "UNIQUE_WATCHLIST_SYNC_ENTRY_SEASON" UNIQUE ("entryId", "seasonNumber"), CONSTRAINT "FK_d9d29444df017aa0cbfa5a2244d" FOREIGN KEY ("entryId") REFERENCES "watchlist_sync_entry" ("id") ON DELETE CASCADE ON UPDATE NO ACTION)` + ); + await queryRunner.query( + `INSERT INTO "temporary_watchlist_sync_entry_season"("id", "seasonNumber", "entryId") SELECT "id", "seasonNumber", "entryId" FROM "watchlist_sync_entry_season"` + ); + await queryRunner.query(`DROP TABLE "watchlist_sync_entry_season"`); + await queryRunner.query( + `ALTER TABLE "temporary_watchlist_sync_entry_season" RENAME TO "watchlist_sync_entry_season"` + ); + await queryRunner.query( + `CREATE INDEX "IDX_d9d29444df017aa0cbfa5a2244" ON "watchlist_sync_entry_season" ("entryId") ` + ); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(`DROP INDEX "IDX_d9d29444df017aa0cbfa5a2244"`); + await queryRunner.query( + `ALTER TABLE "watchlist_sync_entry_season" RENAME TO "temporary_watchlist_sync_entry_season"` + ); + await queryRunner.query( + `CREATE TABLE "watchlist_sync_entry_season" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "seasonNumber" integer NOT NULL, "entryId" integer NOT NULL, CONSTRAINT "UNIQUE_WATCHLIST_SYNC_ENTRY_SEASON" UNIQUE ("entryId", "seasonNumber"))` + ); + await queryRunner.query( + `INSERT INTO "watchlist_sync_entry_season"("id", "seasonNumber", "entryId") SELECT "id", "seasonNumber", "entryId" FROM "temporary_watchlist_sync_entry_season"` + ); + await queryRunner.query( + `DROP TABLE "temporary_watchlist_sync_entry_season"` + ); + await queryRunner.query( + `CREATE INDEX "IDX_d9d29444df017aa0cbfa5a2244" ON "watchlist_sync_entry_season" ("entryId") ` + ); + await queryRunner.query(`DROP INDEX "IDX_acad2425a838db492a20431e32"`); + await queryRunner.query(`DROP INDEX "IDX_76036f4009dac0f28dd3913012"`); + await queryRunner.query( + `ALTER TABLE "watchlist_sync_entry" RENAME TO "temporary_watchlist_sync_entry"` + ); + await queryRunner.query( + `CREATE TABLE "watchlist_sync_entry" ("id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "tmdbId" integer NOT NULL, "mediaType" varchar NOT NULL, "watchlistedAt" integer NOT NULL, "createdAt" datetime NOT NULL DEFAULT (CURRENT_TIMESTAMP), "updatedAt" datetime NOT NULL DEFAULT (CURRENT_TIMESTAMP), "requestedById" integer NOT NULL, CONSTRAINT "UNIQUE_WATCHLIST_SYNC_ENTRY" UNIQUE ("requestedById", "tmdbId", "mediaType"))` + ); + await queryRunner.query( + `INSERT INTO "watchlist_sync_entry"("id", "tmdbId", "mediaType", "watchlistedAt", "createdAt", "updatedAt", "requestedById") SELECT "id", "tmdbId", "mediaType", "watchlistedAt", "createdAt", "updatedAt", "requestedById" FROM "temporary_watchlist_sync_entry"` + ); + await queryRunner.query(`DROP TABLE "temporary_watchlist_sync_entry"`); + await queryRunner.query( + `CREATE INDEX "IDX_acad2425a838db492a20431e32" ON "watchlist_sync_entry" ("requestedById") ` + ); + await queryRunner.query( + `CREATE INDEX "IDX_76036f4009dac0f28dd3913012" ON "watchlist_sync_entry" ("tmdbId") ` + ); + await queryRunner.query(`DROP INDEX "IDX_d9d29444df017aa0cbfa5a2244"`); + await queryRunner.query(`DROP TABLE "watchlist_sync_entry_season"`); + await queryRunner.query(`DROP INDEX "IDX_acad2425a838db492a20431e32"`); + await queryRunner.query(`DROP INDEX "IDX_76036f4009dac0f28dd3913012"`); + await queryRunner.query(`DROP TABLE "watchlist_sync_entry"`); + } +}