fix(portals): serialize Xtream restore revisions

libs/services/src/lib/playlist-backup.service.pins.spec.ts

@@ -217,6 +217,33 @@ describe('PlaylistBackupService Xtream source pins', () => {
         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);

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<void>
+        ) => {
+            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().mockReturnValue(true),
-        },
+        pendingRestoreService,
         ...overrides,
     });
 

libs/services/src/lib/playlist-backup.service.ts

@@ -841,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;
@@ -851,8 +859,14 @@ export class PlaylistBackupService {
             return;
         }
 
-        await this.applyXtreamRestoreState(playlistId, restoreState);
-        if (!this.pendingRestoreService.clear(playlistId, restoreState)) {
+        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.`
             );

libs/services/src/lib/playlist-backup.service.xtream-restore.spec.ts

@@ -86,6 +86,19 @@ 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',
             expect.objectContaining({

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<void>
+        ) => {
+            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().mockReturnValue(true),
-        },
+        pendingRestoreService,
     };
 }

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';
@@ -39,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,
@@ -49,6 +61,7 @@ describe('XtreamPendingRestoreService', () => {
             playbackPositions: [],
         });
 
+        expect(snapshot).not.toBeNull();
         const persisted = JSON.parse(
             localStorage.getItem(storageKey) ?? 'null'
         );
@@ -152,4 +165,222 @@ describe('XtreamPendingRestoreService', () => {
         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<void>((resolve) => {
+            releaseOlder = resolve;
+        });
+        let signalOlderStarted!: () => void;
+        const olderStarted = new Promise<void>((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<void>((resolve) => {
+            releaseFirst = resolve;
+        });
+        let signalFirstStarted!: () => void;
+        const firstStarted = new Promise<void>((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<void>((resolve) => {
+            releaseOlder = resolve;
+        });
+        let signalOlderStarted!: () => void;
+        const olderStarted = new Promise<void>((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();
+    });
 });

libs/services/src/lib/xtream-pending-restore.service.ts

@@ -5,10 +5,32 @@ 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<void>
+    >();
+    private readonly restoreRevisions = new Map<
+        string,
+        { revision: number; serializedState: string }
+    >();
+    private readonly lastConsumedRevisions = new Map<string, number>();
+    private nextRestoreRevision = 0;
+
     get(playlistId: string): XtreamPendingRestoreState | null {
         try {
             return this.getOrThrow(playlistId);
@@ -44,19 +66,119 @@ 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,
+        };
+    }
+
+    set(
+        playlistId: string,
+        state: XtreamPendingRestoreState
+    ): XtreamPendingRestoreSnapshot | null {
+        if (!playlistId) {
+            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<void>
+    ): Promise<XtreamPendingRestoreApplicationResult> {
+        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(
@@ -77,10 +199,7 @@ export class XtreamPendingRestoreService {
 
             if (
                 !currentState ||
-                JSON.stringify(currentState) !==
-                    JSON.stringify(
-                        normalizeXtreamPendingRestoreState(expectedState)
-                    )
+                !this.restoreStatesEqual(currentState, expectedState)
             ) {
                 return false;
             }
@@ -102,12 +221,46 @@ export class XtreamPendingRestoreService {
         try {
             localStorage.removeItem(storageKey);
             if (localStorage.getItem(storageKey) === null) {
+                this.restoreRevisions.delete(playlistId);
                 return true;
             }
         } catch {
             // 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<T>(
+        playlistId: string,
+        apply: () => Promise<T>
+    ): Promise<T> {
+        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;
+    }
 }