fix(server-storage): a restore merged by a conflict stays parked; first-read tests wait for the read

A restore whose write conflicted with an edit another device made during
it had the conflict replay merge that edit in, and flushed() still said
the lists were taken, so the restore was consumed with lists that were
not the backup's. Both restore paths now compare the stored lists with
the ones they wrote once the server took them, and a difference keeps
the restore parked to run again on a fresh read.

The two first-read tests released the mocked read after a fixed number
of microtask turns, which could come before the read started; they now
wait until it has.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.spec.ts

@@ -1091,6 +1091,60 @@ describe('PwaXtreamDataSource', () => {
             expect(flushed).toHaveBeenCalledWith('p1');
         });
 
+        it('replaces the lists, and runs again when a conflict merged them', async () => {
+            const values = new Map<string, string>([
+                ['xtream-favorites', JSON.stringify({ p1: ['live:1'] })],
+            ]);
+            let merge = false;
+            const storage = {
+                getItem: (key: string) => values.get(key) ?? null,
+                setItem: (key: string, value: string) => {
+                    values.set(key, value);
+                },
+                removeItem: (key: string) => {
+                    values.delete(key);
+                },
+                ready: async () => true,
+                flushed: jest.fn(async () => {
+                    if (merge) {
+                        // Another device added one meanwhile: replayed in.
+                        values.set(
+                            'xtream-favorites',
+                            JSON.stringify({ p1: ['live:9', 'movie:5'] })
+                        );
+                    }
+                    return true;
+                }),
+            };
+            TestBed.resetTestingModule();
+            TestBed.configureTestingModule({
+                providers: [
+                    PwaXtreamDataSource,
+                    { provide: USER_DATA_STORAGE, useValue: storage },
+                    { provide: ParentalLockService, useValue: parentalLock },
+                    { provide: XtreamApiService, useValue: apiService },
+                    { provide: PlaylistsService, useValue: playlistsService },
+                ],
+            });
+            const source = TestBed.inject(PwaXtreamDataSource);
+            const state = {
+                hiddenCategories: [],
+                favorites: [{ xtreamId: 5, contentType: 'movie' }],
+                recentlyViewed: [],
+                playbackPositions: [],
+            } as never;
+
+            await source.restoreUserData('p1', state);
+            expect(JSON.parse(values.get('xtream-favorites') ?? '{}')).toEqual({
+                p1: ['movie:5'],
+            });
+
+            merge = true;
+            await expect(source.restoreUserData('p1', state)).rejects.toThrow(
+                /changed elsewhere/
+            );
+        });
+
         it('replaces nothing while earlier changes to the playlist are unsent', async () => {
             const values = new Map<string, string>([
                 ['xtream-favorites', JSON.stringify({ p1: ['live:1'] })],

libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.ts

@@ -1465,6 +1465,23 @@ export class PwaXtreamDataSource implements IXtreamDataSource {
                 'The server did not take the restored lists; the restore waits.'
             );
         }
+        // A conflict merges an edit another device made meanwhile into the
+        // lists: the restore replaces them, so it runs again on a fresh read.
+        const stored = [
+            this.collectionStorage.getFavoritesFromStorage()[playlistId],
+            this.collectionStorage.getRecentItemsFromStorage()[playlistId],
+            this.getPlaybackPositionsFromStorage()[playlistId],
+        ];
+        const restored = [
+            favorites[playlistId],
+            recentItems[playlistId],
+            playbackPositions[playlistId],
+        ];
+        if (JSON.stringify(stored) !== JSON.stringify(restored)) {
+            throw new Error(
+                'The lists changed elsewhere during the restore; it runs again.'
+            );
+        }
     }
 
     // =========================================================================

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

@@ -539,6 +539,41 @@ describe('PlaylistBackupService Xtream user state in the PWA', () => {
             ).not.toBeNull();
         });
 
+        it('keeps the restore parked when a conflict merged other lists in', async () => {
+            const manifest = await exportSeeded();
+            localStorage.clear();
+            // Taken, but merged with an edit another device made meanwhile.
+            const flushed = jest
+                .fn()
+                .mockResolvedValueOnce(true)
+                .mockImplementation(async () => {
+                    const map = JSON.parse(
+                        localStorage.getItem(KEYS.favorites) ?? '{}'
+                    );
+                    map['xtream-1'] = [...map['xtream-1'], 'live:9'];
+                    localStorage.setItem(KEYS.favorites, JSON.stringify(map));
+                    return true;
+                });
+            const service = createPlaylistBackupService({
+                playlistsService: playlistsService([existingXtreamPlaylist]),
+                pwaXtream: createPwaXtreamBackupState({
+                    getItem: (key) => localStorage.getItem(key),
+                    setItem: (key, value) => localStorage.setItem(key, value),
+                    removeItem: (key) => localStorage.removeItem(key),
+                    ready: async () => true,
+                    flushed,
+                }),
+            });
+
+            const summary = await service.importBackup(xtreamOnly(manifest));
+
+            expect(summary.failed).toBe(1);
+            expect(summary.errors[0]).toContain('changed elsewhere');
+            expect(
+                pendingRestore(service).getOrThrow('xtream-1')
+            ).not.toBeNull();
+        });
+
         it('replaces nothing while earlier changes to the playlist are unsent', async () => {
             const manifest = await exportSeeded();
             localStorage.clear();

libs/services/src/lib/pwa-xtream-backup-state.service.ts

@@ -138,8 +138,9 @@ export class PwaXtreamBackupStateService {
                 'Earlier changes to this playlist are not on the server yet; the restore waits.'
             );
         }
+        let written: Map<UserDataMap, unknown[]>;
         try {
-            this.replaceLists(playlistId, state);
+            written = this.replaceLists(playlistId, state);
         } catch (error) {
             if (!isQuotaExceeded(error)) {
                 try {
@@ -153,22 +154,43 @@ export class PwaXtreamBackupStateService {
         // Server mode: the lists count once the server took them; a refused
         // one keeps the restore parked (titles merged regardless, as above).
         const saved = (await this.storage.flushed?.(playlistId)) !== false;
+        // A conflict merges an edit another device made meanwhile into the
+        // lists: the backup replaces them, so it waits and runs again on a
+        // fresh read.
+        const replaced =
+            saved &&
+            [...written].every(
+                ([name, list]) =>
+                    JSON.stringify(this.listFor(name, playlistId)) ===
+                    JSON.stringify(list)
+            );
         this.mergeSnapshots(playlistId, snapshots);
         if (!saved) {
             throw new Error(
                 'The server did not take the restored lists; the restore waits.'
             );
         }
+        if (!replaced) {
+            throw new Error(
+                'The lists changed elsewhere during the restore; it runs again.'
+            );
+        }
     }
 
+    /** The lists it wrote, by map. */
     private replaceLists(
         playlistId: string,
         state: XtreamPendingRestoreState
-    ): void {
+    ): Map<UserDataMap, unknown[]> {
+        const written = new Map<UserDataMap, unknown[]>();
+        const write = (name: UserDataMap, list: unknown[]) => {
+            this.writeFor(name, playlistId, list);
+            written.set(name, list);
+        };
         const favorites = state.favorites
             .map((item) => typedKey(item.contentType, item.xtreamId))
             .filter((key): key is string => key !== null);
-        this.writeFor('favorites', playlistId, [...new Set(favorites)]);
+        write('favorites', [...new Set(favorites)]);
 
         const recent: StoredMap[] = [];
         for (const item of state.recentlyViewed) {
@@ -181,18 +203,18 @@ export class PwaXtreamBackupStateService {
                 });
             }
         }
-        this.writeFor('recentItems', playlistId, recent);
+        write('recentItems', recent);
 
         const now = new Date().toISOString();
-        this.writeFor(
+        write(
             'playbackPositions',
-            playlistId,
             state.playbackPositions.filter(isPlaybackPosition).map((item) => ({
                 ...item,
                 playlistId,
                 updatedAt: item.updatedAt ?? now,
             }))
         );
+        return written;
     }
 
     /**

libs/services/src/lib/server-storage/server-user-data-adopt.spec.ts

@@ -3,6 +3,16 @@ import { UNCONFIRMED_TTL_MS } from './server-unsent-writes';
 import { ServerUserDataStorage } from './server-user-data-storage';
 import { fakeServer, memoryStorage } from './server-user-data.test-helpers';
 
+/** Lets microtasks run until `condition` holds (a mocked call started). */
+async function until(condition: () => boolean): Promise<void> {
+    for (let turn = 0; turn < 1000 && !condition(); turn += 1) {
+        await Promise.resolve();
+    }
+    if (!condition()) {
+        throw new Error('The awaited call never started');
+    }
+}
+
 describe('ServerUserDataStorage taking over other page loads', () => {
     it("keeps progress this tab saved over a closed tab's older edit", async () => {
         const at = (seconds: number) => [
@@ -222,11 +232,13 @@ describe('ServerUserDataStorage taking over other page loads', () => {
         );
         await a.hydrate();
         let answer = (): void => undefined;
+        let asked = false;
         const b = new ServerUserDataStorage(
             {
                 ...server.api,
                 // Answered with the state of when it was asked.
                 getJson: jest.fn(() => {
+                    asked = true;
                     const old = server.api.getJson();
                     return new Promise((resolve) => {
                         answer = () => resolve(old);
@@ -239,9 +251,8 @@ describe('ServerUserDataStorage taking over other page loads', () => {
             memoryStorage()
         );
         const reading = b.hydrate();
-        for (let turn = 0; turn < 10; turn += 1) {
-            await Promise.resolve();
-        }
+        // Its read is out before the other tab's write lands.
+        await until(() => asked);
         a.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] }));
         await a.flush();
         answer();
@@ -426,10 +437,12 @@ describe('ServerUserDataStorage taking over other page loads', () => {
         a.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] }));
         const sending = a.flush();
         let answer = (): void => undefined;
+        let asked = false;
         const b = new ServerUserDataStorage(
             {
                 ...server.api,
                 getJson: jest.fn(() => {
+                    asked = true;
                     const old = server.api.getJson();
                     return new Promise((resolve) => {
                         answer = () => resolve(old);
@@ -442,9 +455,8 @@ describe('ServerUserDataStorage taking over other page loads', () => {
             memoryStorage()
         );
         const reading = b.hydrate();
-        for (let turn = 0; turn < 10; turn += 1) {
-            await Promise.resolve();
-        }
+        // Its read is out before the other tab's write lands.
+        await until(() => asked);
         // A's write lands (its record is cleared), then B's old answer.
         land();
         await sending;