diff --git a/.changes/portals-atomic-fresh-import-pins.md b/.changes/portals-atomic-fresh-import-pins.md new file mode 100644 index 000000000..9ba031ca1 --- /dev/null +++ b/.changes/portals-atomic-fresh-import-pins.md @@ -0,0 +1,9 @@ +--- +type: fix +area: portals +--- + +Importing an Xtream backup now restores all preferred VOD sources together. If +restoration or cleanup fails, the catalog stays retryable and remains blocked +until the parked state is safely consumed, preventing stale replay from +overwriting newer choices. diff --git a/apps/web/src/app/settings/settings-backup.facade.spec.ts b/apps/web/src/app/settings/settings-backup.facade.spec.ts index e4e70ea9d..a14201475 100644 --- a/apps/web/src/app/settings/settings-backup.facade.spec.ts +++ b/apps/web/src/app/settings/settings-backup.facade.spec.ts @@ -4,6 +4,7 @@ import { selectAllPlaylistsMeta, selectIsEpgAvailable, } from '@iptvnator/m3u-state'; +import { XtreamStore } from '@iptvnator/portal/xtream/data-access'; import { PlaylistBackupService } from '@iptvnator/services'; import { provideMockStore } from '@ngrx/store/testing'; import { TranslateModule } from '@ngx-translate/core'; @@ -36,6 +37,9 @@ describe('SettingsBackupFacade', () => { .mockResolvedValue(BACKUP_EXPORT_RESULT), importBackup: jest.fn(), }), + MockProvider(XtreamStore, { + reconcilePendingRestoreBlock: jest.fn(), + }), provideMockStore({ selectors: [ { selector: selectAllPlaylistsMeta, value: [] }, @@ -154,4 +158,46 @@ describe('SettingsBackupFacade', () => { expect.objectContaining({ panelClass: ['settings-snackbar'] }) ); }); + + it('reconciles a pending Xtream restore before completing an import', async () => { + configure(); + const input = document.createElement('input'); + const file = { + text: jest.fn().mockResolvedValue('{}'), + } as unknown as File; + Object.defineProperty(input, 'files', { value: [file] }); + const addEventListener = jest.spyOn(input, 'addEventListener'); + const createElement = jest + .spyOn(document, 'createElement') + .mockReturnValue(input); + jest.spyOn(input, 'click').mockImplementation(); + const summary = { + imported: 0, + merged: 0, + skipped: 0, + failed: 1, + errors: ['pending restore state could not be consumed'], + }; + (playlistBackupService.importBackup as jest.Mock).mockResolvedValue( + summary + ); + const onImported = jest.fn(); + const xtreamStore = TestBed.inject(XtreamStore); + + facade.importData(onImported); + const changeListener = addEventListener.mock.calls.find( + ([type]) => type === 'change' + )?.[1] as (event: Event) => Promise; + await changeListener({ target: input } as unknown as Event); + + expect(xtreamStore.reconcilePendingRestoreBlock).toHaveBeenCalledTimes( + 1 + ); + expect(onImported).toHaveBeenCalledTimes(1); + expect( + (xtreamStore.reconcilePendingRestoreBlock as jest.Mock).mock + .invocationCallOrder[0] + ).toBeLessThan(onImported.mock.invocationCallOrder[0]); + createElement.mockRestore(); + }); }); diff --git a/apps/web/src/app/settings/settings-backup.facade.ts b/apps/web/src/app/settings/settings-backup.facade.ts index 6aa377b4c..bebfc6004 100644 --- a/apps/web/src/app/settings/settings-backup.facade.ts +++ b/apps/web/src/app/settings/settings-backup.facade.ts @@ -1,7 +1,8 @@ -import { inject, Injectable, signal } from '@angular/core'; +import { inject, Injectable, Injector, signal } from '@angular/core'; import { Store } from '@ngrx/store'; import { TranslateService } from '@ngx-translate/core'; import { PlaylistActions } from '@iptvnator/m3u-state'; +import { XtreamStore } from '@iptvnator/portal/xtream/data-access'; import { PlaylistBackupImportSummary, PlaylistBackupService, @@ -16,6 +17,7 @@ export class SettingsBackupFacade { private readonly settingsSnackbar = inject(SettingsSnackbarService); private readonly store = inject(Store); private readonly translate = inject(TranslateService); + private readonly injector = inject(Injector); readonly isExportingData = signal(false); @@ -79,6 +81,9 @@ export class SettingsBackupFacade { const summary = await this.playlistBackupService.importBackup( await file.text() ); + this.injector + .get(XtreamStore, null) + ?.reconcilePendingRestoreBlock(); if (summary.imported > 0 || summary.merged > 0) { this.store.dispatch(PlaylistActions.removeAllPlaylists()); diff --git a/docs/architecture/vod-multi-source.md b/docs/architecture/vod-multi-source.md index d33efbace..16c205a26 100644 --- a/docs/architecture/vod-multi-source.md +++ b/docs/architecture/vod-multi-source.md @@ -673,6 +673,35 @@ optional `sourcePins` collection. See sanitizing rules — the short version is that `matchKey` names the film and survives as-is, while the playlist id becomes the imported copy's. +Both restore routes use the same atomic replacement: the direct backup restore +and the parked replay after a fresh Xtream import call +`VodSourcePinService.replaceForPlaylist`, whose worker operation clears the +playlist's pins and inserts the complete restored set in one transaction. For +direct restore, that preserves the existing pins if an insert fails. A fresh +import has no pre-existing pins to lose; there it prevents a partially applied +prefix from being visible before the parked state is retried. + +The parked state is consumed only after that atomic replacement succeeds, and +its removal is verified (with an empty tombstone as the safe fallback when +storage removal fails). Consumption compares the current parked snapshot with +the one that was applied. Store-owned replay and direct backup restore share a +playlist-scoped FIFO coordinator and consume a revision captured for the +content generation they restored. Duplicate consumers of that revision +coalesce, while a newer revision remains parked for its own post-import +consumer; an older asynchronous replacement therefore cannot clear or finish +after a newer one. Parking is verified before an import can report success, and +whole backup imports are serialized as well. Either restore path reports a +failed import instead of claiming success when consumption cannot be +confirmed. Fresh-import initialization stays blocked and retryable on either +failure. Cached content is not exposed while parked state exists: a complete +active initialization must restore and retire the snapshot before the catalog +opens, because a scope-limited offline cache may not contain every identity +referenced by the backup. Route bootstrap and Settings import completion +reconcile that gate before content can be edited. The active content route also +stays unmounted while replay is pending, and the import overlay follows the +full initialization session (including cache-only retries), not only +remote-download events. + ## Which engines can fail over Only the built-in web players (HTML5, Video.js, ArtPlayer) raise the playback diff --git a/libs/playlist/shared/ui/src/lib/playlist-refresh-action.service.ts b/libs/playlist/shared/ui/src/lib/playlist-refresh-action.service.ts index 214cc380d..dd911f832 100644 --- a/libs/playlist/shared/ui/src/lib/playlist-refresh-action.service.ts +++ b/libs/playlist/shared/ui/src/lib/playlist-refresh-action.service.ts @@ -158,10 +158,16 @@ export class PlaylistRefreshActionService { ] ); - this.pendingRestoreService.set(item._id, { - ...restoreState, - playbackPositions, - }); + if ( + !this.pendingRestoreService.set(item._id, { + ...restoreState, + playbackPositions, + }) + ) { + throw new Error( + `Parking pending restore state for "${item._id}" failed.` + ); + } measureRendererPerformancePhase( RENDERER_PERFORMANCE_PHASE.XTREAM_REFRESH_META, diff --git a/libs/playlist/shared/ui/src/lib/recent-playlists/recent-playlists.component.ts b/libs/playlist/shared/ui/src/lib/recent-playlists/recent-playlists.component.ts index 45d88f585..5cb08a43b 100644 --- a/libs/playlist/shared/ui/src/lib/recent-playlists/recent-playlists.component.ts +++ b/libs/playlist/shared/ui/src/lib/recent-playlists/recent-playlists.component.ts @@ -402,10 +402,16 @@ export class RecentPlaylistsComponent { ] ); - this.pendingRestoreService.set(item._id, { - ...restoreState, - playbackPositions, - }); + if ( + !this.pendingRestoreService.set(item._id, { + ...restoreState, + playbackPositions, + }) + ) { + throw new Error( + `Parking pending restore state for "${item._id}" failed.` + ); + } // Update the timestamp in NgRx / IndexedDB measureRendererPerformancePhase( diff --git a/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.test-helpers.ts b/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.test-helpers.ts index 3c33d89ff..5a1005632 100644 --- a/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.test-helpers.ts +++ b/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.test-helpers.ts @@ -33,7 +33,7 @@ export function createDbServiceMock() { getXtreamCategories: jest.fn().mockResolvedValue([]), saveXtreamCategories: jest.fn().mockResolvedValue(undefined), getAllXtreamCategories: jest.fn().mockResolvedValue([]), - updateCategoryVisibility: jest.fn().mockResolvedValue(undefined), + updateCategoryVisibility: jest.fn().mockResolvedValue(true), hasXtreamContent: jest.fn().mockResolvedValue(false), getXtreamContent: jest.fn().mockResolvedValue([]), saveXtreamContent: jest.fn().mockResolvedValue(0), @@ -60,6 +60,7 @@ export function createDbServiceMock() { export function createVodSourcePinServiceMock() { return { listForPlaylist: jest.fn().mockResolvedValue([]), + replaceForPlaylist: jest.fn().mockResolvedValue(true), set: jest.fn().mockResolvedValue(true), }; } diff --git a/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.ts b/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.ts index d417cc333..7dac4bdcc 100644 --- a/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.ts +++ b/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.ts @@ -566,6 +566,48 @@ export class ElectronXtreamDataSource implements IXtreamDataSource { restoreState: XtreamPendingRestoreState, options?: XtreamOperationOptions ): Promise { + const categoriesByType = await Promise.all([ + this.dbService.getAllXtreamCategories(playlistId, 'live'), + this.dbService.getAllXtreamCategories(playlistId, 'movies'), + this.dbService.getAllXtreamCategories(playlistId, 'series'), + ]); + for (const categories of categoriesByType) { + if (categories.length === 0) { + continue; + } + + const reset = await this.dbService.updateCategoryVisibility( + categories.map((category) => category.id), + false + ); + if (!reset) { + throw new Error( + `Resetting category visibility for "${playlistId}" failed.` + ); + } + + const hiddenCategoryIds = categories + .filter((category) => + restoreState.hiddenCategories.some( + (hiddenCategory) => + hiddenCategory.categoryType === category.type && + hiddenCategory.xtreamId === category.xtream_id + ) + ) + .map((category) => category.id); + if ( + hiddenCategoryIds.length > 0 && + !(await this.dbService.updateCategoryVisibility( + hiddenCategoryIds, + true + )) + ) { + throw new Error( + `Restoring category visibility for "${playlistId}" failed.` + ); + } + } + await this.dbService.restoreXtreamUserData( playlistId, restoreState.favorites, @@ -585,23 +627,29 @@ export class ElectronXtreamDataSource implements IXtreamDataSource { // The fresh-import path lands here rather than in the backup service: // a new playlist has no content yet when the archive is read, so its // user state is parked and applied once the import finishes. - for (const pin of restoreState.sourcePins ?? []) { - const written = await this.vodSourcePinService.set({ - matchKey: pin.matchKey, - playlistId, - contentId: pin.contentId, - portalType: 'xtream', - ...(pin.updatedAt ? { updatedAt: pin.updatedAt } : {}), - }); + if (!restoreState.sourcePins) { + return; + } - // Throwing keeps the pending state for a later retry — the caller - // only clears it when this resolves. Dropping it here would lose - // the preference with the import still reporting success. - if (!written) { - throw new Error( - `Restoring the pinned source for "${pin.matchKey}" failed.` - ); - } + const pins = restoreState.sourcePins.map((pin) => ({ + matchKey: pin.matchKey, + playlistId, + contentId: pin.contentId, + portalType: 'xtream' as const, + ...(pin.updatedAt ? { updatedAt: pin.updatedAt } : {}), + })); + const replaced = await this.vodSourcePinService.replaceForPlaylist( + playlistId, + pins + ); + + // Throwing keeps the pending state for a later retry — the caller only + // clears it when this resolves. Dropping it here would lose the + // preference with the import still reporting success. + if (!replaced) { + throw new Error( + `Restoring the pinned sources for "${playlistId}" failed.` + ); } } } diff --git a/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.user-data.spec.ts b/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.user-data.spec.ts index af578ea28..bddcda14d 100644 --- a/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.user-data.spec.ts +++ b/libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.user-data.spec.ts @@ -198,7 +198,7 @@ describe('ElectronXtreamDataSource (user data delegation)', () => { }); }); - it('applies parked source pins against the imported playlist', async () => { + it('atomically applies all parked source pins and rejects a failed replacement', async () => { // The fresh-import path: a new playlist has no content when the // archive is read, so its user state is parked and replayed here. // Without this the backup's pins are dropped for every new import. @@ -213,48 +213,76 @@ describe('ElectronXtreamDataSource (user data delegation)', () => { contentId: 501, updatedAt: '2026-07-06T09:00:00.000Z', }, + { + matchKey: 'title:the-matrix:1999', + contentId: 502, + }, ], } as never; await harness.dataSource.restoreUserData(playlistId, restoreState); - expect(harness.vodSourcePinService.set).toHaveBeenCalledWith({ - matchKey: 'tmdb:603', - playlistId, - contentId: 501, - portalType: 'xtream', - updatedAt: '2026-07-06T09:00:00.000Z', - }); - }); - - it('keeps the pending state when a pin cannot be written', async () => { - harness.vodSourcePinService.set.mockResolvedValue(false); - const restoreState = { - hiddenCategories: [], - favorites: [], - recentlyViewed: [], - playbackPositions: [], - sourcePins: [{ matchKey: 'tmdb:603', contentId: 501 }], - } as never; + expect( + harness.vodSourcePinService.replaceForPlaylist + ).toHaveBeenCalledTimes(1); + expect( + harness.vodSourcePinService.replaceForPlaylist + ).toHaveBeenCalledWith(playlistId, [ + { + matchKey: 'tmdb:603', + playlistId, + contentId: 501, + portalType: 'xtream', + updatedAt: '2026-07-06T09:00:00.000Z', + }, + { + matchKey: 'title:the-matrix:1999', + playlistId, + contentId: 502, + portalType: 'xtream', + }, + ]); + expect(harness.vodSourcePinService.set).not.toHaveBeenCalled(); // The caller clears the parked state only when this resolves, so - // resolving here would drop the preference on a transient DB - // failure while the import still reported success. + // resolving here would drop the retry on a transient DB failure. + harness.vodSourcePinService.replaceForPlaylist.mockResolvedValue( + false + ); await expect( harness.dataSource.restoreUserData(playlistId, restoreState) - ).rejects.toThrow('tmdb:603'); + ).rejects.toThrow(playlistId); }); it('restores user data, then resets and replays playback positions', async () => { const positionA = { contentXtreamId: 1 } as never; const positionB = { contentXtreamId: 2 } as never; const restoreState = { - hiddenCategories: [], + hiddenCategories: [{ categoryType: 'live', xtreamId: 101 }], favorites: [{ xtreamId: 202, type: 'movie' }], recentlyViewed: [{ xtreamId: 101, type: 'live' }], playbackPositions: [positionA, positionB], } as never; const options = { operationId: 'op-1' }; + harness.dbService.getAllXtreamCategories.mockImplementation( + (_playlistId: string, type: string) => + Promise.resolve( + type === 'live' + ? [ + { + id: 11, + type: 'live', + xtream_id: 101, + }, + { + id: 12, + type: 'live', + xtream_id: 102, + }, + ] + : [] + ) + ); await harness.dataSource.restoreUserData( playlistId, @@ -270,6 +298,12 @@ describe('ElectronXtreamDataSource (user data delegation)', () => { [{ xtreamId: 101, type: 'live' }], options ); + expect( + harness.dbService.updateCategoryVisibility.mock.calls + ).toEqual([ + [[11, 12], false], + [[11], true], + ]); expect( harness.playbackService.clearAllPlaybackPositions ).toHaveBeenCalledWith(playlistId); diff --git a/libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.spec-helpers.ts b/libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.spec-helpers.ts new file mode 100644 index 000000000..8f4d2c204 --- /dev/null +++ b/libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.spec-helpers.ts @@ -0,0 +1,171 @@ +import { patchState, signalStore, withMethods, withState } from '@ngrx/signals'; +import { XtreamPendingRestoreState } from '@iptvnator/shared/interfaces'; +import { + RENDERER_PERFORMANCE_PHASE_HOOK_KEY, + type RendererPerformancePhaseEvent, +} from '@iptvnator/shared/logging'; +import { XtreamPendingRestoreSnapshot } from '@iptvnator/services'; +import { XtreamPlaylistData } from '../../data-sources/xtream-data-source.interface'; +import { PortalStatusType } from '../../xtream-state'; +import { withContent } from './with-content.feature'; + +export const TEST_PLAYLIST: XtreamPlaylistData = { + id: 'playlist-1', + name: 'Test Xtream', + serverUrl: 'http://localhost:8080', + username: 'demo', + password: 'secret', + type: 'xtream', +}; + +export const PENDING_RESTORE_STATE: XtreamPendingRestoreState = { + hiddenCategories: [], + favorites: [{ contentType: 'live', xtreamId: 77 }], + recentlyViewed: [], + playbackPositions: [], + sourcePins: [{ matchKey: 'tmdb:603', contentId: 501 }], +}; + +export function createPendingRestoreServiceMock() { + let nextRevision = 0; + let activeSnapshot: XtreamPendingRestoreSnapshot | null = null; + let activeSerializedState: string | null = null; + let lastConsumedRevision: number | null = null; + const mock = { + getOrThrow: jest.fn().mockReturnValue(null), + getSnapshotOrThrow: jest.fn(), + clear: jest.fn().mockReturnValue(true), + applyAndConsume: jest.fn(), + }; + mock.getSnapshotOrThrow.mockImplementation((playlistId: string) => { + const state = mock.getOrThrow( + playlistId + ) as XtreamPendingRestoreState | null; + if (!state) { + activeSnapshot = null; + activeSerializedState = null; + return null; + } + + const serializedState = JSON.stringify(state); + if ( + !activeSnapshot || + activeSerializedState !== serializedState + ) { + activeSnapshot = { + playlistId, + revision: ++nextRevision, + state, + }; + activeSerializedState = serializedState; + lastConsumedRevision = null; + } + return activeSnapshot; + }); + mock.applyAndConsume.mockImplementation( + async ( + playlistId: string, + expectedSnapshot: XtreamPendingRestoreSnapshot, + apply: (state: XtreamPendingRestoreState) => Promise + ) => { + const currentSnapshot = + mock.getSnapshotOrThrow(playlistId); + if (!currentSnapshot) { + return lastConsumedRevision === expectedSnapshot.revision + ? 'consumed' + : 'superseded'; + } + if ( + currentSnapshot.revision !== expectedSnapshot.revision + ) { + return 'superseded'; + } + + await apply(currentSnapshot.state); + if (!mock.clear(playlistId, currentSnapshot.state)) { + return 'consume-failed'; + } + lastConsumedRevision = currentSnapshot.revision; + activeSnapshot = null; + activeSerializedState = null; + mock.getOrThrow.mockReturnValue(null); + return 'consumed'; + } + ); + return mock; +} + +export function createContentTestStore( + checkPortalStatus: () => Promise +) { + return signalStore( + withState({ + playlistId: TEST_PLAYLIST.id, + currentPlaylist: TEST_PLAYLIST, + portalStatus: 'active' as PortalStatusType, + selectedContentType: 'vod' as const, + }), + withMethods((store) => ({ + async checkPortalStatus(): Promise { + const status = await checkPortalStatus(); + patchState(store, { portalStatus: status }); + return status; + }, + })), + withContent() + ); +} + +const performanceHookSymbol = Symbol.for(RENDERER_PERFORMANCE_PHASE_HOOK_KEY); + +export function setPerformanceHook( + hook: ((event: RendererPerformancePhaseEvent) => void) | null +): void { + const target = globalThis as unknown as Record; + if (hook === null) { + delete target[performanceHookSymbol]; + } else { + target[performanceHookSymbol] = hook; + } +} + +export function readPendingRestoreBlocked(store: unknown): boolean { + return ( + store as { + isPendingRestoreBlocked: () => boolean; + } + ).isPendingRestoreBlocked(); +} + +export function createDeferred() { + let resolve!: (value: T) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + + return { promise, resolve, reject }; +} + +export function createAbortError(): Error { + const error = new Error('Import cancelled'); + error.name = 'AbortError'; + return error; +} + +export async function waitForCondition( + predicate: () => boolean, + attempts = 20 +): Promise { + for (let index = 0; index < attempts; index += 1) { + if (predicate()) { + return; + } + + await Promise.resolve(); + await new Promise((resolve) => setTimeout(resolve, 0)); + } + + throw new Error('Timed out waiting for test condition'); +} diff --git a/libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.spec.ts b/libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.spec.ts index d210d259e..8acb45a06 100644 --- a/libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.spec.ts +++ b/libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.spec.ts @@ -1,17 +1,22 @@ import { TestBed } from '@angular/core/testing'; -import { patchState, signalStore, withMethods, withState } from '@ngrx/signals'; -import { DatabaseService } from '@iptvnator/services'; import { - RENDERER_PERFORMANCE_PHASE_HOOK_KEY, - type RendererPerformancePhaseEvent, -} from '@iptvnator/shared/logging'; -import { - XTREAM_DATA_SOURCE, - XtreamPlaylistData, -} from '../../data-sources/xtream-data-source.interface'; + DatabaseService, + XtreamPendingRestoreService, +} from '@iptvnator/services'; +import { XTREAM_DATA_SOURCE } from '../../data-sources/xtream-data-source.interface'; import { XtreamApiService } from '../../services/xtream-api.service'; import { PortalStatusType } from '../../xtream-state'; -import { withContent } from './with-content.feature'; +import { + createAbortError, + createContentTestStore, + createDeferred, + createPendingRestoreServiceMock, + PENDING_RESTORE_STATE, + readPendingRestoreBlocked, + setPerformanceHook, + TEST_PLAYLIST, + waitForCondition, +} from './with-content.feature.spec-helpers'; jest.mock('@iptvnator/portal/shared/util', () => ({ createLogger: () => ({ @@ -24,78 +29,11 @@ jest.mock('@iptvnator/portal/shared/util', () => ({ type ContentType = 'live' | 'movie' | 'series'; -const PLAYLIST: XtreamPlaylistData = { - id: 'playlist-1', - name: 'Test Xtream', - serverUrl: 'http://localhost:8080', - username: 'demo', - password: 'secret', - type: 'xtream', -}; -const performanceHookSymbol = Symbol.for(RENDERER_PERFORMANCE_PHASE_HOOK_KEY); - -function setPerformanceHook( - hook: ((event: RendererPerformancePhaseEvent) => void) | null -): void { - const target = globalThis as unknown as Record; - if (hook === null) { - delete target[performanceHookSymbol]; - } else { - target[performanceHookSymbol] = hook; - } -} +const PLAYLIST = TEST_PLAYLIST; let checkPortalStatusMock: jest.Mock, []>; -const TestContentStore = signalStore( - withState({ - playlistId: PLAYLIST.id, - currentPlaylist: PLAYLIST, - portalStatus: 'active' as PortalStatusType, - selectedContentType: 'vod' as const, - }), - withMethods((store) => ({ - async checkPortalStatus(): Promise { - const status = await checkPortalStatusMock(); - patchState(store, { portalStatus: status }); - return status; - }, - })), - withContent() -); - -function createDeferred() { - let resolve!: (value: T) => void; - let reject!: (reason?: unknown) => void; - const promise = new Promise((res, rej) => { - resolve = res; - reject = rej; - }); - - return { promise, resolve, reject }; -} - -function createAbortError(): Error { - const error = new Error('Import cancelled'); - error.name = 'AbortError'; - return error; -} - -async function waitForCondition( - predicate: () => boolean, - attempts = 20 -): Promise { - for (let index = 0; index < attempts; index += 1) { - if (predicate()) { - return; - } - - await Promise.resolve(); - await new Promise((resolve) => setTimeout(resolve, 0)); - } - - throw new Error('Timed out waiting for test condition'); -} +const TestContentStore = createContentTestStore(() => checkPortalStatusMock()); describe('withContent import state', () => { let store: InstanceType; @@ -119,6 +57,9 @@ describe('withContent import state', () => { let xtreamApiService: { cancelSession: jest.Mock; }; + let pendingRestoreService: ReturnType< + typeof createPendingRestoreServiceMock + >; beforeEach(() => { localStorage.clear(); @@ -153,6 +94,7 @@ describe('withContent import state', () => { xtreamApiService = { cancelSession: jest.fn().mockResolvedValue(true), }; + pendingRestoreService = createPendingRestoreServiceMock(); checkPortalStatusMock = jest.fn().mockResolvedValue('active'); TestBed.configureTestingModule({ @@ -170,6 +112,10 @@ describe('withContent import state', () => { provide: XtreamApiService, useValue: xtreamApiService, }, + { + provide: XtreamPendingRestoreService, + useValue: pendingRestoreService, + }, ], }); @@ -614,6 +560,63 @@ describe('withContent import state', () => { expect(databaseService.getXtreamImportStatus).not.toHaveBeenCalled(); }); + it('does not expose offline cache while a restore is pending', async () => { + pendingRestoreService.getOrThrow.mockReturnValue(PENDING_RESTORE_STATE); + dataSource.hasCategories.mockResolvedValue(true); + dataSource.hasContent.mockResolvedValue(true); + + await expect(store.hasUsableOfflineCache('vod')).resolves.toBe(false); + await store.hydrateCachedContent('vod'); + + expect(dataSource.hasCategories).not.toHaveBeenCalled(); + expect(dataSource.hasContent).not.toHaveBeenCalled(); + expect(dataSource.getCachedCategories).not.toHaveBeenCalled(); + expect(dataSource.getCachedContent).not.toHaveBeenCalled(); + expect(dataSource.restoreUserData).not.toHaveBeenCalled(); + expect(store.isContentInitialized()).toBe(false); + expect(store.contentInitBlockReason()).toBe('error'); + }); + + it('blocks active initialization when pending restore storage is unreadable', async () => { + dataSource.getContent.mockResolvedValue([]); + pendingRestoreService.getOrThrow.mockImplementation(() => { + throw new Error('storage is locked'); + }); + + await store.initializeContent(); + + expect(dataSource.restoreUserData).not.toHaveBeenCalled(); + expect(store.isContentInitialized()).toBe(false); + expect(store.contentInitBlockReason()).toBe('error'); + }); + + it('fails closed when pending restore storage is unreadable', async () => { + pendingRestoreService.getOrThrow.mockImplementation(() => { + throw new Error('storage is locked'); + }); + dataSource.hasCategories.mockResolvedValue(true); + dataSource.hasContent.mockResolvedValue(true); + + await expect(store.hasUsableOfflineCache('vod')).resolves.toBe(false); + await store.hydrateCachedContent('vod'); + + expect(dataSource.hasCategories).not.toHaveBeenCalled(); + expect(dataSource.hasContent).not.toHaveBeenCalled(); + expect(dataSource.getCachedContent).not.toHaveBeenCalled(); + expect(store.isContentInitialized()).toBe(false); + expect(store.contentInitBlockReason()).toBe('error'); + + pendingRestoreService.getOrThrow.mockReturnValue(null); + checkPortalStatusMock.mockResolvedValue('unavailable'); + dataSource.hasContent.mockResolvedValue(true); + await store.retryContentInitialization(); + + expect(readPendingRestoreBlocked(store)).toBe(false); + expect(dataSource.getCachedContent).toHaveBeenCalled(); + expect(store.isContentInitialized()).toBe(true); + expect(store.contentInitBlockReason()).toBeNull(); + }); + it('does not require every content type for aggregate cached sections', async () => { dataSource.hasContent .mockResolvedValueOnce(false) @@ -738,6 +741,33 @@ describe('withContent import state', () => { expect(store.vodStreams()).toHaveLength(1); }); + it('blocks cache publishing when pending state appears during hydration', async () => { + const cachedCategories = createDeferred(); + const cachedContent = createDeferred(); + dataSource.getCachedCategories.mockReturnValueOnce( + cachedCategories.promise + ); + dataSource.getCachedContent.mockReturnValueOnce(cachedContent.promise); + + const hydration = store.hydrateCachedContent('vod'); + await waitForCondition( + () => dataSource.getCachedContent.mock.calls.length === 1 + ); + + pendingRestoreService.getOrThrow.mockReturnValue(PENDING_RESTORE_STATE); + cachedCategories.resolve([]); + cachedContent.resolve([]); + + await hydration; + + expect(dataSource.restoreUserData).not.toHaveBeenCalled(); + expect(pendingRestoreService.clear).not.toHaveBeenCalled(); + expect(store.vodStreams()).toEqual([]); + expect(store.isContentInitialized()).toBe(false); + expect(store.isCachedContentScopeReady('vod')).toBe(false); + expect(store.contentInitBlockReason()).toBe('error'); + }); + it('coalesces concurrent cached hydration calls for the same scope', async () => { const cachedCategories = createDeferred(); const cachedContent = createDeferred(); @@ -991,6 +1021,145 @@ describe('withContent import state', () => { expect(dataSource.getContent).toHaveBeenCalledTimes(3); }); + it('keeps a failed pending restore blocked until an explicit retry succeeds', async () => { + dataSource.getContent.mockResolvedValue([]); + dataSource.restoreUserData + .mockRejectedValueOnce(new Error('database is locked')) + .mockResolvedValueOnce(undefined); + pendingRestoreService.getOrThrow.mockReturnValue(PENDING_RESTORE_STATE); + pendingRestoreService.clear.mockImplementation(() => { + pendingRestoreService.getOrThrow.mockReturnValue(null); + return true; + }); + + await store.initializeContent(); + + expect(dataSource.restoreUserData).toHaveBeenCalledTimes(1); + expect(pendingRestoreService.clear).not.toHaveBeenCalled(); + expect(store.isContentInitialized()).toBe(false); + expect(store.contentInitBlockReason()).toBe('error'); + + await store.initializeContent(); + expect(dataSource.restoreUserData).toHaveBeenCalledTimes(1); + + // Ready cache rows are not safe to expose while replay is pending: + // a later full replacement could otherwise erase pins changed in the + // meantime. + expect(store.isCachedContentScopeReady('vod')).toBe(false); + checkPortalStatusMock.mockResolvedValue('unavailable'); + dataSource.hasContent.mockResolvedValue(true); + await store.retryContentInitialization(); + + expect(dataSource.restoreUserData).toHaveBeenCalledTimes(1); + expect(store.isContentInitialized()).toBe(false); + expect(store.contentInitBlockReason()).toBe('unavailable'); + + checkPortalStatusMock.mockResolvedValue('active'); + await store.retryContentInitialization(); + + expect(dataSource.restoreUserData).toHaveBeenCalledTimes(2); + expect(pendingRestoreService.clear).toHaveBeenCalledWith( + PLAYLIST.id, + PENDING_RESTORE_STATE + ); + expect(store.isContentInitialized()).toBe(true); + expect(store.contentInitBlockReason()).toBeNull(); + }); + + it('leaves state parked during an active import for the next content generation', async () => { + const pendingSeries = createDeferred(); + dataSource.getContent.mockImplementation( + (_playlistId: string, _credentials: unknown, type: ContentType) => + type === 'series' ? pendingSeries.promise : Promise.resolve([]) + ); + + const initialization = store.initializeContent(); + await waitForCondition( + () => store.contentLoadStateByType().vod === 'ready' + ); + + pendingRestoreService.getOrThrow.mockReturnValue(PENDING_RESTORE_STATE); + expect(store.reconcilePendingRestoreBlock()).toBe(true); + expect(readPendingRestoreBlocked(store)).toBe(true); + expect(store.isImporting()).toBe(false); + expect(store.activeImportSessionId()).not.toBeNull(); + expect(store.isContentInitialized()).toBe(false); + expect(dataSource.restoreUserData).not.toHaveBeenCalled(); + + pendingSeries.resolve([]); + await initialization; + + expect(dataSource.restoreUserData).not.toHaveBeenCalled(); + expect(readPendingRestoreBlocked(store)).toBe(true); + expect(store.isContentInitialized()).toBe(false); + + dataSource.getContent.mockResolvedValue([]); + await store.initializeContent(); + + expect(dataSource.restoreUserData).toHaveBeenCalledWith( + PLAYLIST.id, + PENDING_RESTORE_STATE, + expect.any(Object) + ); + expect(pendingRestoreService.applyAndConsume).toHaveBeenCalledWith( + PLAYLIST.id, + expect.objectContaining({ + state: PENDING_RESTORE_STATE, + }), + expect.any(Function) + ); + expect(pendingRestoreService.clear).toHaveBeenCalledWith( + PLAYLIST.id, + PENDING_RESTORE_STATE + ); + expect(readPendingRestoreBlocked(store)).toBe(false); + expect(store.isContentInitialized()).toBe(true); + }); + + it('retries when pending state could not be consumed', async () => { + dataSource.getContent.mockResolvedValue([]); + pendingRestoreService.getOrThrow.mockReturnValue(PENDING_RESTORE_STATE); + pendingRestoreService.clear + .mockReturnValueOnce(false) + .mockImplementationOnce(() => { + pendingRestoreService.getOrThrow.mockReturnValue(null); + return true; + }); + + await store.initializeContent(); + + expect(dataSource.restoreUserData).toHaveBeenCalledTimes(1); + expect(store.isContentInitialized()).toBe(false); + expect(store.contentInitBlockReason()).toBe('error'); + + await store.retryContentInitialization(); + + expect(dataSource.restoreUserData).toHaveBeenCalledTimes(2); + expect(pendingRestoreService.clear).toHaveBeenCalledTimes(2); + expect(store.isContentInitialized()).toBe(true); + expect(store.contentInitBlockReason()).toBeNull(); + }); + + it('keeps cancellation authoritative while a pending restore finishes', async () => { + const pendingRestore = createDeferred(); + dataSource.getContent.mockResolvedValue([]); + dataSource.restoreUserData.mockReturnValueOnce(pendingRestore.promise); + pendingRestoreService.getOrThrow.mockReturnValue(PENDING_RESTORE_STATE); + + const initialization = store.initializeContent(); + await waitForCondition( + () => dataSource.restoreUserData.mock.calls.length === 1 + ); + + await store.cancelImport(); + pendingRestore.reject(new Error('pin replacement failed')); + await expect(initialization).resolves.toBeUndefined(); + + expect(pendingRestoreService.clear).not.toHaveBeenCalled(); + expect(store.isContentInitialized()).toBe(false); + expect(store.contentInitBlockReason()).toBe('cancelled'); + }); + it('stops before content fetch if cancel lands between categories and content phases', async () => { const pendingCategories = { live: createDeferred(), diff --git a/libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.ts b/libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.ts index 15ac0f490..4dbe1492d 100644 --- a/libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.ts +++ b/libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.ts @@ -106,6 +106,7 @@ export interface ContentState { activeImportSessionId: string | null; activeImportOperationIds: string[]; isContentInitialized: boolean; + isPendingRestoreBlocked: boolean; contentInitBlockReason: XtreamContentInitBlockReason | null; } @@ -141,6 +142,7 @@ const initialContentState: ContentState = { activeImportSessionId: null, activeImportOperationIds: [], isContentInitialized: false, + isPendingRestoreBlocked: false, contentInitBlockReason: null, }; @@ -270,6 +272,18 @@ export function withContent() { const asCachedContent = (content: unknown): T[] => content as T[]; + const hasPendingRestoreOrReadFailure = ( + playlistId: string + ): boolean => { + try { + return ( + pendingRestoreService.getOrThrow(playlistId) !== null + ); + } catch { + return true; + } + }; + const markContentScopeLoading = ( scope?: XtreamCachedContentScope | null, options?: { preserveInitialized?: boolean } @@ -404,6 +418,10 @@ export function withContent() { playlistId: string, scope?: XtreamCachedContentScope | null ): Promise => { + if (hasPendingRestoreOrReadFailure(playlistId)) { + return false; + } + const types = getTypesForCacheScope(scope); if ( @@ -419,10 +437,16 @@ export function withContent() { ) ) ); - return checks.some(Boolean); + return ( + checks.some(Boolean) && + !hasPendingRestoreOrReadFailure(playlistId) + ); } - return hasCachedContentForType(playlistId, scope); + return ( + (await hasCachedContentForType(playlistId, scope)) && + !hasPendingRestoreOrReadFailure(playlistId) + ); }; const isCurrentCachedHydrationContext = ( @@ -451,6 +475,35 @@ export function withContent() { return types.every((type) => loadStates[type] === 'ready'); }; + const blockCacheForPendingRestore = ( + playlistId: string, + scope?: XtreamCachedContentScope | null + ): boolean => { + if (!hasPendingRestoreOrReadFailure(playlistId)) { + return false; + } + + patchState(store, (state) => { + const nextLoadStates = { + ...state.contentLoadStateByType, + }; + for (const type of getTypesForCacheScope(scope)) { + nextLoadStates[type] = 'error'; + } + + return { + isLoadingCategories: false, + isLoadingContent: false, + isContentInitialized: false, + isPendingRestoreBlocked: true, + contentInitBlockReason: + state.contentInitBlockReason ?? 'error', + contentLoadStateByType: nextLoadStates, + }; + }); + return true; + }; + const executeCachedContentHydration = async ( playlistId: string, scope: XtreamCachedContentScope | null | undefined, @@ -520,6 +573,10 @@ export function withContent() { return; } + if (blockCacheForPendingRestore(playlistId, scope)) { + return; + } + patchState(store, (state) => { const nextLoadStates = { ...state.contentLoadStateByType, @@ -529,6 +586,7 @@ export function withContent() { isLoadingContent: false, isImporting: false, isContentInitialized: true, + isPendingRestoreBlocked: false, contentInitBlockReason: null, }; @@ -573,11 +631,16 @@ export function withContent() { const ctx = getCredentialsFromStore(); if (!ctx) return; + if (blockCacheForPendingRestore(ctx.playlistId, scope)) { + return; + } + if (isCachedContentScopeReady(scope)) { patchState(store, { isLoadingCategories: false, isLoadingContent: false, isContentInitialized: true, + isPendingRestoreBlocked: false, contentInitBlockReason: null, }); return; @@ -788,6 +851,19 @@ export function withContent() { const completedTypes = new Set(); try { + // Capture parked state before publishing any imported + // content. A retry may load each type from the DB without + // emitting an import phase, so the store-owned gate is what + // prevents source-pin edits until replay is consumed. + const initialRestoreSnapshot = + pendingRestoreService.getSnapshotOrThrow( + ctx.playlistId + ); + patchState(store, { + isPendingRestoreBlocked: + initialRestoreSnapshot !== null, + }); + // Electron content persistence maps remote category IDs // to internal DB category rows, so categories must exist // before content import starts. @@ -802,11 +878,21 @@ export function withContent() { }); throwIfImportCancelled(importSessionId); + // Only the revision captured before import belongs to this + // content generation. A newer revision may come from a + // refresh that has deleted the catalog and must remain + // parked until its own replacement import completes. + const currentRestoreSnapshot = + pendingRestoreService.getSnapshotOrThrow( + ctx.playlistId + ); + patchState(store, { + isPendingRestoreBlocked: + currentRestoreSnapshot !== null, + }); + // Restore user data if needed - const restoreData = pendingRestoreService.get( - ctx.playlistId - ); - if (restoreData) { + if (initialRestoreSnapshot) { try { throwIfImportCancelled(importSessionId); const restoreOperationId = @@ -817,23 +903,49 @@ export function withContent() { patchState(store, { importPhase: 'restoring-favorites', }); - await dataSource.restoreUserData( - ctx.playlistId, - restoreData, - { - onEvent: trackImportEvent, - operationId: restoreOperationId, - } - ); - throwIfImportCancelled(importSessionId); - pendingRestoreService.clear(ctx.playlistId); + const restoreResult = + await pendingRestoreService.applyAndConsume( + ctx.playlistId, + initialRestoreSnapshot, + async (pendingState) => { + throwIfImportCancelled(importSessionId); + await dataSource.restoreUserData( + ctx.playlistId, + pendingState, + { + onEvent: trackImportEvent, + operationId: + restoreOperationId, + } + ); + throwIfImportCancelled(importSessionId); + } + ); + if (restoreResult === 'consume-failed') { + throw new Error( + `Clearing pending restore state for "${ctx.playlistId}" failed.` + ); + } } catch (err) { + throwIfImportCancelled(importSessionId); + if (!isDbAbortError(err)) { logger.error('Error restoring user data', err); } + + throw err; } } + const isRestoreStillPending = + hasPendingRestoreOrReadFailure(ctx.playlistId); + patchState(store, { + isPendingRestoreBlocked: isRestoreStillPending, + }); + if (isRestoreStillPending) { + return; + } + throwIfImportCancelled(importSessionId); // Mark as initialized so next routings won't re-trigger it @@ -1183,6 +1295,25 @@ export function withContent() { await runContentInitialization(); }, + reconcilePendingRestoreBlock(): boolean { + const ctx = getCredentialsFromStore(); + if (!ctx) { + return false; + } + + const isBlocked = hasPendingRestoreOrReadFailure( + ctx.playlistId + ); + patchState(store, (state) => ({ + isPendingRestoreBlocked: isBlocked, + contentInitBlockReason: + isBlocked && !state.activeImportSessionId + ? (state.contentInitBlockReason ?? 'error') + : state.contentInitBlockReason, + })); + return isBlocked; + }, + async hasUsableOfflineCache( scope?: XtreamCachedContentScope | null ): Promise { @@ -1203,7 +1334,12 @@ export function withContent() { isCachedContentScopeReady( scope?: XtreamCachedContentScope | null ): boolean { - return isCachedContentScopeReady(scope); + const ctx = getCredentialsFromStore(); + return ( + (!ctx || + !hasPendingRestoreOrReadFailure(ctx.playlistId)) && + isCachedContentScopeReady(scope) + ); }, async hydrateCachedContent( diff --git a/libs/portal/xtream/data-access/src/lib/xtream-state.ts b/libs/portal/xtream/data-access/src/lib/xtream-state.ts index 700a8dddf..4c3becfd2 100644 --- a/libs/portal/xtream/data-access/src/lib/xtream-state.ts +++ b/libs/portal/xtream/data-access/src/lib/xtream-state.ts @@ -65,6 +65,7 @@ export interface XtreamState { epgItems: EpgItem[]; hideExternalInfoDialog: boolean; portalStatus: PortalStatusType; + isPendingRestoreBlocked: boolean; contentInitBlockReason: XtreamContentInitBlockReason | null; globalSearchResults: GlobalSearchResult[]; streamUrl: string; diff --git a/libs/portal/xtream/feature/src/lib/xtream-content-gate.component.spec.ts b/libs/portal/xtream/feature/src/lib/xtream-content-gate.component.spec.ts index 30ae79211..255833334 100644 --- a/libs/portal/xtream/feature/src/lib/xtream-content-gate.component.spec.ts +++ b/libs/portal/xtream/feature/src/lib/xtream-content-gate.component.spec.ts @@ -32,6 +32,7 @@ describe('XtreamContentGateComponent', () => { const contentInitBlockReason = signal(null); const isContentInitialized = signal(false); + const isPendingRestoreBlocked = signal(false); const portalStatus = signal<'active' | 'inactive' | 'expired' | 'unavailable'>( 'active' ); @@ -40,6 +41,7 @@ describe('XtreamContentGateComponent', () => { beforeEach(async () => { contentInitBlockReason.set(null); isContentInitialized.set(false); + isPendingRestoreBlocked.set(false); portalStatus.set('active'); retryContentInitialization.mockClear(); @@ -65,6 +67,7 @@ describe('XtreamContentGateComponent', () => { useValue: { contentInitBlockReason, isContentInitialized, + isPendingRestoreBlocked, portalStatus, retryContentInitialization, }, @@ -110,10 +113,19 @@ describe('XtreamContentGateComponent', () => { it('keeps the child outlet available when there is no block reason', () => { fixture.detectChanges(); + expect(fixture.nativeElement.querySelector('.mock-error')).toBeNull(); + expect(fixture.nativeElement.querySelector('router-outlet')).not.toBeNull(); + }); + + it('keeps the child outlet unavailable while parked state is pending', () => { + isPendingRestoreBlocked.set(true); + fixture.detectChanges(); + + expect(fixture.nativeElement.querySelector('router-outlet')).toBeNull(); expect( fixture.nativeElement.querySelector('.mock-error') - ).toBeNull(); - expect(fixture.nativeElement.querySelector('router-outlet')).not.toBeNull(); + ).not.toBeNull(); + expect(fixture.nativeElement.querySelector('button')).not.toBeNull(); }); it('shows an inline warning when cached content remains available offline', () => { diff --git a/libs/portal/xtream/feature/src/lib/xtream-content-gate.component.ts b/libs/portal/xtream/feature/src/lib/xtream-content-gate.component.ts index e53f35e12..421f08fe2 100644 --- a/libs/portal/xtream/feature/src/lib/xtream-content-gate.component.ts +++ b/libs/portal/xtream/feature/src/lib/xtream-content-gate.component.ts @@ -24,7 +24,7 @@ import { XtreamCachedOfflineNoticeComponent } from './xtream-cached-offline-noti XtreamCachedOfflineNoticeComponent, ], template: ` - @if (contentInitBlockReason(); as blockReason) { + @if (effectiveBlockReason(); as blockReason) {
+ this.contentInitBlockReason() ?? + (this.isPendingRestoreBlocked() ? 'error' : null) + ); private readonly errorViewKey = computed(() => { - switch (this.contentInitBlockReason()) { + switch (this.effectiveBlockReason()) { case 'cancelled': return 'IMPORT_CANCELLED'; case 'expired': diff --git a/libs/portal/xtream/feature/src/lib/xtream-workspace-route-session.service.spec.ts b/libs/portal/xtream/feature/src/lib/xtream-workspace-route-session.service.spec.ts index a8aca7734..fb4525532 100644 --- a/libs/portal/xtream/feature/src/lib/xtream-workspace-route-session.service.spec.ts +++ b/libs/portal/xtream/feature/src/lib/xtream-workspace-route-session.service.spec.ts @@ -116,6 +116,7 @@ describe('XtreamWorkspaceRouteSession', () => { setCurrentPlaylist: jest.fn((playlist: XtreamPlaylistData | null) => { currentPlaylist.set(playlist); }), + reconcilePendingRestoreBlock: jest.fn().mockReturnValue(false), fetchXtreamPlaylist: jest.fn().mockResolvedValue(undefined), checkPortalStatus: jest.fn(), hasUsableOfflineCache: jest.fn().mockImplementation(async () => { @@ -236,6 +237,7 @@ describe('XtreamWorkspaceRouteSession', () => { xtreamStore.resetStore.mockClear(); xtreamStore.setCurrentPlaylist.mockClear(); + xtreamStore.reconcilePendingRestoreBlock.mockClear(); xtreamStore.fetchXtreamPlaylist.mockClear(); xtreamStore.checkPortalStatus.mockReset(); xtreamStore.hasUsableOfflineCache.mockClear(); @@ -361,6 +363,16 @@ describe('XtreamWorkspaceRouteSession', () => { await flushEffects(); expect(xtreamStore.resetStore).toHaveBeenCalledWith(PLAYLIST_ID); + expect( + xtreamStore.reconcilePendingRestoreBlock.mock.invocationCallOrder[0] + ).toBeGreaterThan( + xtreamStore.setCurrentPlaylist.mock.invocationCallOrder[0] + ); + expect( + xtreamStore.reconcilePendingRestoreBlock.mock.invocationCallOrder[0] + ).toBeLessThan( + xtreamStore.fetchXtreamPlaylist.mock.invocationCallOrder[0] + ); expect(xtreamStore.setSelectedContentType).toHaveBeenCalledWith('live'); expect(xtreamStore.prepareContentLoading).toHaveBeenCalledWith('live'); expect(selectedContentType()).toBe('live'); diff --git a/libs/portal/xtream/feature/src/lib/xtream-workspace-route-session.service.ts b/libs/portal/xtream/feature/src/lib/xtream-workspace-route-session.service.ts index 90aba19d2..0b48352a7 100644 --- a/libs/portal/xtream/feature/src/lib/xtream-workspace-route-session.service.ts +++ b/libs/portal/xtream/feature/src/lib/xtream-workspace-route-session.service.ts @@ -298,6 +298,7 @@ export class XtreamWorkspaceRouteSession { didBootstrapPlaylist = true; this.xtreamStore.setCurrentPlaylist(routePlaylist); + this.xtreamStore.reconcilePendingRestoreBlock(); section = this.syncRouteState(routeSection); if (isImportDrivenSection(section)) { this.xtreamStore.prepareContentLoading(cacheScope); diff --git a/libs/services/src/lib/playlist-backup.service.pins.spec.ts b/libs/services/src/lib/playlist-backup.service.pins.spec.ts index 98eea4d2c..ea7c7efca 100644 --- a/libs/services/src/lib/playlist-backup.service.pins.spec.ts +++ b/libs/services/src/lib/playlist-backup.service.pins.spec.ts @@ -116,6 +116,56 @@ describe('PlaylistBackupService Xtream source pins', () => { ]); }); + it('serializes overlapping restores so an older snapshot cannot finish last', async () => { + const collaborators = createRestoreCollaborators(); + let finishFirstReplacement!: (replaced: boolean) => void; + const firstReplacement = new Promise((resolve) => { + finishFirstReplacement = resolve; + }); + const replacePins = jest + .fn() + .mockReturnValueOnce(firstReplacement) + .mockResolvedValueOnce(true); + const service = createPlaylistBackupService({ + ...collaborators, + vodSourcePinService: { + isAvailable: true, + listForPlaylistOrThrow: jest.fn().mockResolvedValue([]), + replaceForPlaylist: replacePins, + }, + }); + const firstImport = service.importBackup( + JSON.stringify( + createXtreamManifest( + [], + [{ matchKey: 'tmdb:603', contentId: 501 }] + ) + ) + ); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(replacePins).toHaveBeenCalledTimes(1); + + const secondImport = service.importBackup( + JSON.stringify( + createXtreamManifest( + [], + [{ matchKey: 'tmdb:603', contentId: 502 }] + ) + ) + ); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(replacePins).toHaveBeenCalledTimes(1); + + finishFirstReplacement(true); + await Promise.all([firstImport, secondImport]); + expect( + replacePins.mock.calls.map( + ([, pins]) => (pins as Array<{ contentId: number }>)[0].contentId + ) + ).toEqual([501, 502]); + }); + it('reports pins that could not be written', async () => { const collaborators = createRestoreCollaborators(); const service = createPlaylistBackupService({ @@ -142,6 +192,58 @@ describe('PlaylistBackupService Xtream source pins', () => { ); }); + it('reports a restore whose parked state could not be consumed', async () => { + const collaborators = createRestoreCollaborators(); + collaborators.pendingRestoreService.clear.mockReturnValue(false); + const service = createPlaylistBackupService({ + ...collaborators, + vodSourcePinService: { + isAvailable: true, + listForPlaylistOrThrow: jest.fn().mockResolvedValue([]), + replaceForPlaylist: jest.fn().mockResolvedValue(true), + }, + }); + + const manifest = createXtreamManifest( + [], + [{ matchKey: 'tmdb:603', contentId: 501 }] + ); + + const summary = await service.importBackup(JSON.stringify(manifest)); + + expect(summary).toEqual( + expect.objectContaining({ merged: 0, failed: 1 }) + ); + expect(summary.errors[0]).toMatch(/pending restore state/i); + }); + + it('reports a restore whose pending state could not be parked', async () => { + const collaborators = createRestoreCollaborators(); + collaborators.pendingRestoreService.set.mockReturnValueOnce(null); + const replacePins = jest.fn().mockResolvedValue(true); + const service = createPlaylistBackupService({ + ...collaborators, + vodSourcePinService: { + isAvailable: true, + listForPlaylistOrThrow: jest.fn().mockResolvedValue([]), + replaceForPlaylist: replacePins, + }, + }); + + const manifest = createXtreamManifest( + [], + [{ matchKey: 'tmdb:603', contentId: 501 }] + ); + + const summary = await service.importBackup(JSON.stringify(manifest)); + + expect(summary).toEqual( + expect.objectContaining({ merged: 0, failed: 1 }) + ); + expect(summary.errors[0]).toMatch(/parking pending restore state/i); + expect(replacePins).not.toHaveBeenCalled(); + }); + it('drops pins the backup does not contain', async () => { const collaborators = createRestoreCollaborators(); const clearPins = jest.fn().mockResolvedValue(true); diff --git a/libs/services/src/lib/playlist-backup.service.test-helpers.ts b/libs/services/src/lib/playlist-backup.service.test-helpers.ts index a14f9f3af..684380ab6 100644 --- a/libs/services/src/lib/playlist-backup.service.test-helpers.ts +++ b/libs/services/src/lib/playlist-backup.service.test-helpers.ts @@ -5,9 +5,13 @@ import { VodSourcePin, XtreamBackupFavoriteItem, XtreamBackupRecentlyViewedItem, + XtreamPendingRestoreState, } from '@iptvnator/shared/interfaces'; import { PlaylistBackupService } from './playlist-backup.service'; -import { XtreamPendingRestoreService } from './xtream-pending-restore.service'; +import { + XtreamPendingRestoreService, + XtreamPendingRestoreSnapshot, +} from './xtream-pending-restore.service'; /** * Shared factory for PlaylistBackupService specs. Instantiates the service @@ -20,6 +24,58 @@ export function createPlaylistBackupService( const service = Object.create( PlaylistBackupService.prototype ) as PlaylistBackupService; + let nextRevision = 0; + let pendingSnapshot: XtreamPendingRestoreSnapshot | null = null; + let lastConsumedRevision: number | null = null; + const pendingRestoreService = { + set: jest.fn( + ( + playlistId: string, + state: XtreamPendingRestoreState + ): XtreamPendingRestoreSnapshot | null => { + pendingSnapshot = { + playlistId, + revision: ++nextRevision, + state, + }; + lastConsumedRevision = null; + return pendingSnapshot; + } + ), + clear: jest.fn().mockReturnValue(true), + applyAndConsume: jest.fn(), + }; + pendingRestoreService.applyAndConsume.mockImplementation( + async ( + playlistId: string, + expectedSnapshot: XtreamPendingRestoreSnapshot, + apply: (state: XtreamPendingRestoreState) => Promise + ) => { + if (!pendingSnapshot) { + return lastConsumedRevision === expectedSnapshot.revision + ? 'consumed' + : 'superseded'; + } + if ( + pendingSnapshot.revision !== expectedSnapshot.revision + ) { + return 'superseded'; + } + + await apply(pendingSnapshot.state); + if ( + !pendingRestoreService.clear( + playlistId, + pendingSnapshot.state + ) + ) { + return 'consume-failed'; + } + lastConsumedRevision = pendingSnapshot.revision; + pendingSnapshot = null; + return 'consumed'; + } + ); Object.assign(service as object, { playlistsService: { @@ -68,10 +124,7 @@ export function createPlaylistBackupService( clear: jest.fn().mockResolvedValue(true), clearForPlaylist: jest.fn().mockResolvedValue(true), }, - pendingRestoreService: { - set: jest.fn(), - clear: jest.fn(), - }, + pendingRestoreService, ...overrides, }); diff --git a/libs/services/src/lib/playlist-backup.service.ts b/libs/services/src/lib/playlist-backup.service.ts index 881041f5f..479490003 100644 --- a/libs/services/src/lib/playlist-backup.service.ts +++ b/libs/services/src/lib/playlist-backup.service.ts @@ -65,6 +65,7 @@ export class PlaylistBackupService { private readonly pendingRestoreService = inject( XtreamPendingRestoreService ); + private backupImportTail?: Promise; async exportBackup(): Promise { const playlists = await firstValueFrom( @@ -94,7 +95,20 @@ export class PlaylistBackupService { }; } - async importBackup(json: string): Promise { + importBackup(json: string): Promise { + const importResult = ( + this.backupImportTail ?? Promise.resolve() + ).then(() => this.executeImportBackup(json)); + this.backupImportTail = importResult.then( + () => undefined, + () => undefined + ); + return importResult; + } + + private async executeImportBackup( + json: string + ): Promise { const manifest = this.parseManifest(json); const existingPlaylists = await firstValueFrom( this.playlistsService.getAllData() @@ -827,7 +841,15 @@ export class PlaylistBackupService { const restoreState: XtreamPendingRestoreState = normalizeXtreamPendingRestoreState(entry.userState); - this.pendingRestoreService.set(playlistId, restoreState); + const restoreSnapshot = this.pendingRestoreService.set( + playlistId, + restoreState + ); + if (!restoreSnapshot) { + throw new PlaylistBackupError( + `Parking pending restore state for "${playlistId}" failed.` + ); + } if (!this.hasElectronApi()) { return; @@ -837,8 +859,18 @@ export class PlaylistBackupService { return; } - await this.applyXtreamRestoreState(playlistId, restoreState); - this.pendingRestoreService.clear(playlistId); + const restoreResult = + await this.pendingRestoreService.applyAndConsume( + playlistId, + restoreSnapshot, + (pendingState) => + this.applyXtreamRestoreState(playlistId, pendingState) + ); + if (restoreResult !== 'consumed') { + throw new PlaylistBackupError( + `Clearing pending restore state for "${playlistId}" failed.` + ); + } } private async hasCompletedOfflineCache( diff --git a/libs/services/src/lib/playlist-backup.service.xtream-restore.spec.ts b/libs/services/src/lib/playlist-backup.service.xtream-restore.spec.ts index 61f8fe308..a708a7566 100644 --- a/libs/services/src/lib/playlist-backup.service.xtream-restore.spec.ts +++ b/libs/services/src/lib/playlist-backup.service.xtream-restore.spec.ts @@ -86,8 +86,24 @@ describe('PlaylistBackupService Xtream hidden categories (issue #1017)', () => { expect( collaborators.databaseService.updateCategoryVisibility ).toHaveBeenCalledTimes(3); + expect( + collaborators.pendingRestoreService.applyAndConsume + ).toHaveBeenCalledWith( + 'xtream-1', + expect.objectContaining({ + state: expect.objectContaining({ + hiddenCategories: [ + { categoryType: 'live', xtreamId: 101 }, + ], + }), + }), + expect.any(Function) + ); expect(collaborators.pendingRestoreService.clear).toHaveBeenCalledWith( - 'xtream-1' + 'xtream-1', + expect.objectContaining({ + hiddenCategories: [{ categoryType: 'live', xtreamId: 101 }], + }) ); }); diff --git a/libs/services/src/lib/playlist-backup.xtream-fixtures.ts b/libs/services/src/lib/playlist-backup.xtream-fixtures.ts index 7e7e95004..17de2a304 100644 --- a/libs/services/src/lib/playlist-backup.xtream-fixtures.ts +++ b/libs/services/src/lib/playlist-backup.xtream-fixtures.ts @@ -4,8 +4,10 @@ import { PlaylistBackupManifestV1, PLAYLIST_BACKUP_KIND, PLAYLIST_BACKUP_VERSION, + XtreamPendingRestoreState, XtreamPlaylistBackupEntry, } from '@iptvnator/shared/interfaces'; +import { XtreamPendingRestoreSnapshot } from './xtream-pending-restore.service'; /** * The Xtream restore scaffolding both backup suites need — the existing @@ -91,6 +93,59 @@ export function createXtreamManifest( } export function createRestoreCollaborators() { + let nextRevision = 0; + let pendingSnapshot: XtreamPendingRestoreSnapshot | null = null; + let lastConsumedRevision: number | null = null; + const pendingRestoreService = { + set: jest.fn( + ( + playlistId: string, + state: XtreamPendingRestoreState + ): XtreamPendingRestoreSnapshot | null => { + pendingSnapshot = { + playlistId, + revision: ++nextRevision, + state, + }; + lastConsumedRevision = null; + return pendingSnapshot; + } + ), + clear: jest.fn().mockReturnValue(true), + applyAndConsume: jest.fn(), + }; + pendingRestoreService.applyAndConsume.mockImplementation( + async ( + playlistId: string, + expectedSnapshot: XtreamPendingRestoreSnapshot, + apply: (state: XtreamPendingRestoreState) => Promise + ) => { + if (!pendingSnapshot) { + return lastConsumedRevision === expectedSnapshot.revision + ? 'consumed' + : 'superseded'; + } + if ( + pendingSnapshot.revision !== expectedSnapshot.revision + ) { + return 'superseded'; + } + + await apply(pendingSnapshot.state); + if ( + !pendingRestoreService.clear( + playlistId, + pendingSnapshot.state + ) + ) { + return 'consume-failed'; + } + lastConsumedRevision = pendingSnapshot.revision; + pendingSnapshot = null; + return 'consumed'; + } + ); + return { playlistsService: { addPlaylist: jest.fn((playlist: Playlist) => of(playlist)), @@ -111,9 +166,6 @@ export function createRestoreCollaborators() { restoreXtreamUserData: jest.fn().mockResolvedValue(undefined), updateCategoryVisibility: jest.fn().mockResolvedValue(true), }, - pendingRestoreService: { - set: jest.fn(), - clear: jest.fn(), - }, + pendingRestoreService, }; } diff --git a/libs/services/src/lib/xtream-pending-restore.service.spec.ts b/libs/services/src/lib/xtream-pending-restore.service.spec.ts index 772125fab..43eefb9e7 100644 --- a/libs/services/src/lib/xtream-pending-restore.service.spec.ts +++ b/libs/services/src/lib/xtream-pending-restore.service.spec.ts @@ -1,5 +1,17 @@ import { getXtreamPendingRestoreStorageKey } from '@iptvnator/shared/interfaces'; -import { XtreamPendingRestoreService } from './xtream-pending-restore.service'; +import { + XtreamPendingRestoreService, + XtreamPendingRestoreSnapshot, +} from './xtream-pending-restore.service'; + +function requireSnapshot( + snapshot: XtreamPendingRestoreSnapshot | null +): XtreamPendingRestoreSnapshot { + if (!snapshot) { + throw new Error('Expected pending restore snapshot'); + } + return snapshot; +} describe('XtreamPendingRestoreService', () => { const playlistId = 'playlist-1'; @@ -12,6 +24,7 @@ describe('XtreamPendingRestoreService', () => { }); afterEach(() => { + jest.restoreAllMocks(); localStorage.clear(); }); @@ -38,7 +51,7 @@ describe('XtreamPendingRestoreService', () => { }); it('normalizes state on write', () => { - service.set(playlistId, { + const snapshot = service.set(playlistId, { hiddenCategories: [ { categoryType: 'live', xtreamId: 101 }, { categoryType: 'live' } as never, @@ -48,6 +61,7 @@ describe('XtreamPendingRestoreService', () => { playbackPositions: [], }); + expect(snapshot).not.toBeNull(); const persisted = JSON.parse( localStorage.getItem(storageKey) ?? 'null' ); @@ -62,4 +76,311 @@ describe('XtreamPendingRestoreService', () => { localStorage.setItem(storageKey, '{not json'); expect(service.get(playlistId)).toBeNull(); }); + + it('offers a strict read that reports storage access failures', () => { + jest.spyOn(Storage.prototype, 'getItem').mockImplementation(() => { + throw new Error('storage is locked'); + }); + + expect(service.get(playlistId)).toBeNull(); + expect(() => service.getOrThrow(playlistId)).toThrow( + 'storage is locked' + ); + }); + + it('clears persisted state and reports that it was consumed', () => { + service.set(playlistId, { + hiddenCategories: [], + favorites: [], + recentlyViewed: [], + playbackPositions: [], + }); + + expect(service.clear(playlistId)).toBe(true); + expect(service.get(playlistId)).toBeNull(); + expect(localStorage.getItem(storageKey)).toBeNull(); + }); + + it.each([ + { + failure: 'throws', + remove: () => { + throw new Error('storage is locked'); + }, + }, + { + failure: 'does not remove the key', + remove: () => undefined, + }, + ])( + 'consumes state with a tombstone when removeItem $failure', + ({ remove }) => { + service.set(playlistId, { + hiddenCategories: [], + favorites: [], + recentlyViewed: [], + playbackPositions: [], + }); + jest.spyOn(Storage.prototype, 'removeItem').mockImplementation( + remove + ); + + expect(service.clear(playlistId)).toBe(true); + expect(service.get(playlistId)).toBeNull(); + expect(localStorage.getItem(storageKey)).toBe(''); + } + ); + + it('reports failure when state can neither be tombstoned nor removed', () => { + service.set(playlistId, { + hiddenCategories: [], + favorites: [], + recentlyViewed: [], + playbackPositions: [], + }); + jest.spyOn(Storage.prototype, 'setItem').mockImplementation(() => { + throw new Error('storage is locked'); + }); + jest.spyOn(Storage.prototype, 'removeItem').mockImplementation(() => { + throw new Error('storage is locked'); + }); + + expect(service.clear(playlistId)).toBe(false); + expect(service.get(playlistId)).not.toBeNull(); + }); + + it('does not clear a newer snapshot than the one already restored', () => { + const restoredState = { + hiddenCategories: [], + favorites: [{ xtreamId: 101, contentType: 'live' as const }], + recentlyViewed: [], + playbackPositions: [], + }; + const newerState = { + ...restoredState, + favorites: [{ xtreamId: 202, contentType: 'movie' as const }], + }; + service.set(playlistId, newerState); + + expect(service.clear(playlistId, restoredState)).toBe(false); + expect(service.getOrThrow(playlistId)).toEqual(newerState); + }); + + it('leaves a newer snapshot for its own queued consumer', async () => { + const olderState = { + hiddenCategories: [], + favorites: [{ xtreamId: 101, contentType: 'live' as const }], + recentlyViewed: [], + playbackPositions: [], + }; + const newerState = { + ...olderState, + favorites: [{ xtreamId: 202, contentType: 'movie' as const }], + }; + let releaseOlder!: () => void; + const olderCanFinish = new Promise((resolve) => { + releaseOlder = resolve; + }); + let signalOlderStarted!: () => void; + const olderStarted = new Promise((resolve) => { + signalOlderStarted = resolve; + }); + const appliedSnapshots: number[] = []; + + const olderSnapshot = requireSnapshot( + service.set(playlistId, olderState) + ); + const olderApplication = service.applyAndConsume( + playlistId, + olderSnapshot, + async (state) => { + const xtreamId = state.favorites[0]?.xtreamId; + signalOlderStarted(); + await olderCanFinish; + appliedSnapshots.push(xtreamId); + } + ); + await olderStarted; + + const newerSnapshot = requireSnapshot( + service.set(playlistId, newerState) + ); + const newerApply = jest.fn(async (state) => { + appliedSnapshots.push(state.favorites[0]?.xtreamId); + }); + const newerApplication = service.applyAndConsume( + playlistId, + newerSnapshot, + newerApply + ); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(newerApply).not.toHaveBeenCalled(); + + releaseOlder(); + await expect(olderApplication).resolves.toBe('superseded'); + await expect(newerApplication).resolves.toBe('consumed'); + expect(appliedSnapshots).toEqual([101, 202]); + expect(newerApply).toHaveBeenCalledTimes(1); + expect(service.getOrThrow(playlistId)).toBeNull(); + }); + + it('coalesces concurrent consumers of the same snapshot', async () => { + const state = { + hiddenCategories: [], + favorites: [{ xtreamId: 101, contentType: 'live' as const }], + recentlyViewed: [], + playbackPositions: [], + }; + let releaseFirst!: () => void; + const firstCanFinish = new Promise((resolve) => { + releaseFirst = resolve; + }); + let signalFirstStarted!: () => void; + const firstStarted = new Promise((resolve) => { + signalFirstStarted = resolve; + }); + const firstApply = jest.fn(async () => { + signalFirstStarted(); + await firstCanFinish; + }); + const secondApply = jest.fn().mockResolvedValue(undefined); + + const snapshot = requireSnapshot(service.set(playlistId, state)); + const firstApplication = service.applyAndConsume( + playlistId, + snapshot, + firstApply + ); + await firstStarted; + const secondApplication = service.applyAndConsume( + playlistId, + snapshot, + secondApply + ); + + releaseFirst(); + await expect(firstApplication).resolves.toBe('consumed'); + await expect(secondApplication).resolves.toBe('consumed'); + expect(firstApply).toHaveBeenCalledTimes(1); + expect(secondApply).not.toHaveBeenCalled(); + }); + + it('does not consume an identical snapshot parked by a newer producer', async () => { + const state = { + hiddenCategories: [], + favorites: [{ xtreamId: 101, contentType: 'live' as const }], + recentlyViewed: [], + playbackPositions: [], + }; + let releaseOlder!: () => void; + const olderCanFinish = new Promise((resolve) => { + releaseOlder = resolve; + }); + let signalOlderStarted!: () => void; + const olderStarted = new Promise((resolve) => { + signalOlderStarted = resolve; + }); + const olderSnapshot = requireSnapshot( + service.set(playlistId, state) + ); + const olderApplication = service.applyAndConsume( + playlistId, + olderSnapshot, + async () => { + signalOlderStarted(); + await olderCanFinish; + } + ); + await olderStarted; + + const newerSnapshot = requireSnapshot( + service.set(playlistId, state) + ); + expect(newerSnapshot.revision).not.toBe(olderSnapshot.revision); + releaseOlder(); + + await expect( + olderApplication + ).resolves.toBe('superseded'); + expect(service.getSnapshotOrThrow(playlistId)).toEqual(newerSnapshot); + }); + + it('continues the playlist queue after an application rejects', async () => { + const olderState = { + hiddenCategories: [], + favorites: [{ xtreamId: 101, contentType: 'live' as const }], + recentlyViewed: [], + playbackPositions: [], + }; + const newerState = { + ...olderState, + favorites: [{ xtreamId: 202, contentType: 'movie' as const }], + }; + const olderSnapshot = requireSnapshot( + service.set(playlistId, olderState) + ); + + await expect( + service.applyAndConsume( + playlistId, + olderSnapshot, + async () => { + throw new Error('database is locked'); + } + ) + ).rejects.toThrow('database is locked'); + const newerSnapshot = requireSnapshot( + service.set(playlistId, newerState) + ); + const applyNewer = jest.fn().mockResolvedValue(undefined); + + await expect( + service.applyAndConsume( + playlistId, + newerSnapshot, + applyNewer + ) + ).resolves.toBe('consumed'); + expect(applyNewer).toHaveBeenCalledTimes(1); + }); + + it('reports failure when an applied snapshot cannot be consumed', async () => { + const state = { + hiddenCategories: [], + favorites: [{ xtreamId: 101, contentType: 'live' as const }], + recentlyViewed: [], + playbackPositions: [], + }; + const snapshot = requireSnapshot(service.set(playlistId, state)); + jest.spyOn(Storage.prototype, 'setItem').mockImplementation(() => { + throw new Error('storage is locked'); + }); + jest.spyOn(Storage.prototype, 'removeItem').mockImplementation(() => { + throw new Error('storage is locked'); + }); + const apply = jest.fn().mockResolvedValue(undefined); + + await expect( + service.applyAndConsume(playlistId, snapshot, apply) + ).resolves.toBe('consume-failed'); + expect(apply).toHaveBeenCalledWith(state); + expect(service.getOrThrow(playlistId)).toEqual(state); + }); + + it('reports failure when pending state cannot be parked', () => { + jest.spyOn(Storage.prototype, 'setItem').mockImplementation(() => { + throw new Error('storage is locked'); + }); + + expect( + service.set(playlistId, { + hiddenCategories: [], + favorites: [], + recentlyViewed: [], + playbackPositions: [], + }) + ).toBeNull(); + expect(service.getOrThrow(playlistId)).toBeNull(); + }); }); diff --git a/libs/services/src/lib/xtream-pending-restore.service.ts b/libs/services/src/lib/xtream-pending-restore.service.ts index 59b6178f6..ca568c9df 100644 --- a/libs/services/src/lib/xtream-pending-restore.service.ts +++ b/libs/services/src/lib/xtream-pending-restore.service.ts @@ -5,24 +5,58 @@ import { XtreamPendingRestoreState, } from '@iptvnator/shared/interfaces'; +export interface XtreamPendingRestoreSnapshot { + readonly playlistId: string; + readonly revision: number; + readonly state: XtreamPendingRestoreState; +} + +export type XtreamPendingRestoreApplicationResult = + | 'consumed' + | 'superseded' + | 'consume-failed'; + @Injectable({ providedIn: 'root', }) export class XtreamPendingRestoreService { + private readonly restoreApplicationTails = new Map< + string, + Promise + >(); + private readonly restoreRevisions = new Map< + string, + { revision: number; serializedState: string } + >(); + private readonly lastConsumedRevisions = new Map(); + private nextRestoreRevision = 0; + get(playlistId: string): XtreamPendingRestoreState | null { + try { + return this.getOrThrow(playlistId); + } catch { + return null; + } + } + + /** + * Strict restore-path read. Storage access failures are not equivalent to + * an absent snapshot: callers that could expose mutable content must fail + * closed until they can prove the state is missing or consumed. + */ + getOrThrow(playlistId: string): XtreamPendingRestoreState | null { if (!playlistId) { return null; } + const rawState = localStorage.getItem( + getXtreamPendingRestoreStorageKey(playlistId) + ); + if (!rawState) { + return null; + } + try { - const rawState = localStorage.getItem( - getXtreamPendingRestoreStorageKey(playlistId) - ); - - if (!rawState) { - return null; - } - // Persisted state may predate the current build (e.g. entries // written by versions affected by issue #1017), so it is // re-normalized on every read, not only on write. @@ -32,32 +66,201 @@ export class XtreamPendingRestoreService { } } - set(playlistId: string, state: XtreamPendingRestoreState): void { - if (!playlistId) { - return; + getSnapshotOrThrow( + playlistId: string + ): XtreamPendingRestoreSnapshot | null { + const state = this.getOrThrow(playlistId); + if (!state) { + this.restoreRevisions.delete(playlistId); + return null; } - try { - localStorage.setItem( - getXtreamPendingRestoreStorageKey(playlistId), - JSON.stringify(normalizeXtreamPendingRestoreState(state)) - ); - } catch { - // Ignore local storage write failures. + const serializedState = JSON.stringify(state); + let revision = this.restoreRevisions.get(playlistId); + if (!revision || revision.serializedState !== serializedState) { + revision = { + revision: ++this.nextRestoreRevision, + serializedState, + }; + this.restoreRevisions.set(playlistId, revision); + this.lastConsumedRevisions.delete(playlistId); } + + return { + playlistId, + revision: revision.revision, + state, + }; } - clear(playlistId: string): void { + set( + playlistId: string, + state: XtreamPendingRestoreState + ): XtreamPendingRestoreSnapshot | null { if (!playlistId) { - return; + return null; + } + + const normalizedState = normalizeXtreamPendingRestoreState(state); + const serializedState = JSON.stringify(normalizedState); + + try { + const storageKey = getXtreamPendingRestoreStorageKey(playlistId); + localStorage.setItem(storageKey, serializedState); + if (localStorage.getItem(storageKey) !== serializedState) { + return null; + } + } catch { + return null; + } + + const revision = ++this.nextRestoreRevision; + this.restoreRevisions.set(playlistId, { + revision, + serializedState, + }); + this.lastConsumedRevisions.delete(playlistId); + return { + playlistId, + revision, + state: normalizedState, + }; + } + + /** + * Serializes state-bound restore consumers for one playlist. Consumers of + * the same revision coalesce after the first succeeds, while a newer + * revision remains parked for the importer whose content generation is + * ready to accept it. + */ + async applyAndConsume( + playlistId: string, + expectedSnapshot: XtreamPendingRestoreSnapshot, + apply: (state: XtreamPendingRestoreState) => Promise + ): Promise { + if ( + !playlistId || + expectedSnapshot.playlistId !== playlistId + ) { + return 'consume-failed'; + } + + return this.enqueueRestoreApplication(playlistId, async () => { + const currentSnapshot = this.getSnapshotOrThrow(playlistId); + if (!currentSnapshot) { + return this.lastConsumedRevisions.get(playlistId) === + expectedSnapshot.revision + ? 'consumed' + : 'superseded'; + } + if ( + currentSnapshot.revision !== expectedSnapshot.revision + ) { + return 'superseded'; + } + + await apply(currentSnapshot.state); + + const snapshotAfterApply = + this.getSnapshotOrThrow(playlistId); + if ( + !snapshotAfterApply || + snapshotAfterApply.revision !== expectedSnapshot.revision + ) { + return 'superseded'; + } + if (!this.clear(playlistId, snapshotAfterApply.state)) { + return 'consume-failed'; + } + + this.lastConsumedRevisions.set( + playlistId, + expectedSnapshot.revision + ); + return 'consumed'; + }); + } + + clear( + playlistId: string, + expectedState?: XtreamPendingRestoreState + ): boolean { + if (!playlistId) { + return false; + } + + if (expectedState) { + let currentState: XtreamPendingRestoreState | null; + try { + currentState = this.getOrThrow(playlistId); + } catch { + return false; + } + + if ( + !currentState || + !this.restoreStatesEqual(currentState, expectedState) + ) { + return false; + } + } + + const storageKey = getXtreamPendingRestoreStorageKey(playlistId); + let tombstoneStored = false; + + // An empty value is already treated as "no pending restore" by get(). + // Persist it first so a throwing or no-op remove cannot leave the old + // snapshot eligible for a later destructive replay. + try { + localStorage.setItem(storageKey, ''); + tombstoneStored = localStorage.getItem(storageKey) === ''; + } catch { + // Removal may still work when writing is unavailable. } try { - localStorage.removeItem( - getXtreamPendingRestoreStorageKey(playlistId) - ); + localStorage.removeItem(storageKey); + if (localStorage.getItem(storageKey) === null) { + this.restoreRevisions.delete(playlistId); + return true; + } } catch { - // Ignore local storage remove failures. + // A verified tombstone is sufficient even if removal fails. } + + if (tombstoneStored) { + this.restoreRevisions.delete(playlistId); + } + return tombstoneStored; + } + + private restoreStatesEqual( + left: XtreamPendingRestoreState, + right: XtreamPendingRestoreState + ): boolean { + return ( + JSON.stringify(normalizeXtreamPendingRestoreState(left)) === + JSON.stringify(normalizeXtreamPendingRestoreState(right)) + ); + } + + private enqueueRestoreApplication( + playlistId: string, + apply: () => Promise + ): Promise { + const previous = + this.restoreApplicationTails.get(playlistId) ?? Promise.resolve(); + const result = previous.then(apply); + const tail = result.then( + () => undefined, + () => undefined + ); + this.restoreApplicationTails.set(playlistId, tail); + void tail.then(() => { + if (this.restoreApplicationTails.get(playlistId) === tail) { + this.restoreApplicationTails.delete(playlistId); + } + }); + return result; } } diff --git a/libs/shared/interfaces/src/lib/xtream-restore-state.util.ts b/libs/shared/interfaces/src/lib/xtream-restore-state.util.ts index 6f17b039b..99cbc5af2 100644 --- a/libs/shared/interfaces/src/lib/xtream-restore-state.util.ts +++ b/libs/shared/interfaces/src/lib/xtream-restore-state.util.ts @@ -14,7 +14,8 @@ export interface XtreamPendingRestoreState { playbackPositions: PlaybackPositionData[]; /** * Optional: absent from archives and persisted entries written before - * multi-source existed. The normalizer always fills it. + * multi-source existed. The normalizer preserves that absence so restore + * can distinguish "no opinion" from an authoritative empty collection. */ sourcePins?: XtreamBackupSourcePin[]; } diff --git a/libs/workspace/shell/feature/src/lib/workspace-shell/services/workspace-shell-xtream-import.service.ts b/libs/workspace/shell/feature/src/lib/workspace-shell/services/workspace-shell-xtream-import.service.ts index 7ffc7fe3b..35c64202a 100644 --- a/libs/workspace/shell/feature/src/lib/workspace-shell/services/workspace-shell-xtream-import.service.ts +++ b/libs/workspace/shell/feature/src/lib/workspace-shell/services/workspace-shell-xtream-import.service.ts @@ -67,7 +67,6 @@ export class WorkspaceShellXtreamImportService { readonly canCancelXtreamImport = computed( () => this.isElectron && - this.xtreamStore.isImporting() && Boolean(this.xtreamStore.activeImportSessionId()) && !this.xtreamStore.isCancellingImport() ); @@ -193,7 +192,7 @@ export class WorkspaceShellXtreamImportService { readonly isImportRunning = computed( () => !this.xtreamStore.contentInitBlockReason() && - this.xtreamStore.isImporting() + Boolean(this.xtreamStore.activeImportSessionId()) ); readonly isRefreshPreparationRunning = computed(() => Boolean(this.activeRefreshPreparation()) diff --git a/libs/workspace/shell/feature/src/lib/workspace-shell/services/workspace-shell.facade.spec.ts b/libs/workspace/shell/feature/src/lib/workspace-shell/services/workspace-shell.facade.spec.ts index 06585d642..eaa580238 100644 --- a/libs/workspace/shell/feature/src/lib/workspace-shell/services/workspace-shell.facade.spec.ts +++ b/libs/workspace/shell/feature/src/lib/workspace-shell/services/workspace-shell.facade.spec.ts @@ -444,6 +444,18 @@ describe('WorkspaceShellFacade', () => { expect(facade.showXtreamImportOverlay()).toBe(true); }); + it('keeps a cache-only Xtream import session blocking and cancellable', () => { + const xtreamStore = TestBed.inject( + XtreamStore + ) as unknown as MockXtreamStore; + const xtreamImport = TestBed.inject(WorkspaceShellXtreamImportService); + xtreamStore.activeImportSessionId.set('xtream-import-session'); + + expect(xtreamStore.isImporting()).toBe(false); + expect(facade.showXtreamImportOverlay()).toBe(true); + expect(xtreamImport.canCancelXtreamImport()).toBe(true); + }); + it('shows the Xtream overlay during refresh preparation on the dashboard', () => { facade.currentUrl.set('/workspace/dashboard'); refreshPreparationSignal.set({