diff --git a/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.spec.ts b/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.spec.ts index 28fd338a7..39ba106de 100644 --- a/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.spec.ts +++ b/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.spec.ts @@ -1138,6 +1138,17 @@ describe('PwaXtreamDataSource', () => { expect(JSON.parse(values.get('xtream-favorites') ?? '{}')).toEqual({ p1: ['movie:5'], }); + // A saved restore may list one favorite twice: reads drop the + // copy on both sides of the comparison. + await expect( + source.restoreUserData('p1', { + ...(state as object), + favorites: [ + { xtreamId: 5, contentType: 'movie' }, + { xtreamId: 5, contentType: 'movie' }, + ], + } as never) + ).resolves.toBeUndefined(); merge = true; await expect(source.restoreUserData('p1', state)).rejects.toThrow( diff --git a/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.ts b/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.ts index 9c55b7972..23f77d2c2 100644 --- a/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.ts +++ b/libs/portal/xtream/data-access/src/lib/data-sources/pwa-xtream-data-source.ts @@ -1458,6 +1458,14 @@ export class PwaXtreamDataSource implements IXtreamDataSource { }) ); this.savePlaybackPositionsToStorage(playbackPositions); + // The lists as reads return them (deduplicated, normalized). + const restoredLists = () => + JSON.stringify([ + this.collectionStorage.getFavoritesFromStorage()[playlistId], + this.collectionStorage.getRecentItemsFromStorage()[playlistId], + this.getPlaybackPositionsFromStorage()[playlistId], + ]); + const restored = restoredLists(); // Server mode: done once the server took the lists, else it stays // parked for the next open. if ((await this.userDataStorage.flushed?.(playlistId)) === false) { @@ -1467,17 +1475,7 @@ export class PwaXtreamDataSource implements IXtreamDataSource { } // 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)) { + if (restoredLists() !== restored) { throw new Error( 'The lists changed elsewhere during the restore; it runs again.' ); diff --git a/libs/services/src/lib/server-storage/server-user-data-adopt.spec.ts b/libs/services/src/lib/server-storage/server-user-data-adopt.spec.ts index 249e5f206..7e761988f 100644 --- a/libs/services/src/lib/server-storage/server-user-data-adopt.spec.ts +++ b/libs/services/src/lib/server-storage/server-user-data-adopt.spec.ts @@ -1,5 +1,11 @@ import { ServerApiError } from './server-api-client'; -import { UNCONFIRMED_TTL_MS } from './server-unsent-writes'; +import { keepPages } from './server-page-pointer'; +import { + markClosed, + markUnconfirmed, + saveUnsentWrites, + UNCONFIRMED_TTL_MS, +} from './server-unsent-writes'; import { ServerUserDataStorage } from './server-user-data-storage'; import { fakeServer, memoryStorage } from './server-user-data.test-helpers'; @@ -593,4 +599,87 @@ describe('ServerUserDataStorage taking over other page loads', () => { duplicate.touch(); expect(duplicate.hasUnsentWrites()).toBe(true); }); + + describe('a page load and the older one it kept', () => { + // The older page added X; the newer one took that over and then + // removed X again; neither copy could replace the other. + function journals() { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const local = memoryStorage(); + const tag = `t${JSON.stringify(['S']).replace(/"/g, "'")}`; + const entry = (steps: { before: unknown; after: unknown }[]) => [ + { + namespace: 'xtream-favorites' as const, + playlistId: 'p1', + value: steps[steps.length - 1].after, + tag, + steps, + }, + ]; + saveUnsentWrites( + local, + 'default', + 'older', + entry([{ before: ['S'], after: ['S', 'X'] }]) + ); + saveUnsentWrites( + local, + 'default', + 'newer', + entry([ + { before: ['S'], after: ['S', 'X'] }, + { before: ['S', 'X'], after: ['S'] }, + ]) + ); + // Another device adds Y meanwhile: the replay decides. + server.state['xtream-favorites']['p1'] = ['S', 'Y']; + return { server, local }; + } + + async function send(page: ServerUserDataStorage) { + await page.hydrate(); + await page.flush(); + await page.flush(); + } + + it('replays them oldest first on a reload', async () => { + const { server, local } = journals(); + const tabStore = memoryStorage(); + keepPages(tabStore, 'default', 'newer', ['older']); + + await send( + new ServerUserDataStorage( + server.api as never, + local, + jest.fn(), + 'default', + tabStore + ) + ); + + // X stays removed. + expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'Y']); + }); + + it('replays them oldest first when a duplicate adopts them', async () => { + const { server, local } = journals(); + const tabStore = memoryStorage(); + keepPages(tabStore, 'default', 'newer', ['older']); + // The newer page load still runs when the duplicate opens. + markUnconfirmed(local, 'default', 'newer', ['xtream-favorites|p1']); + const duplicate = new ServerUserDataStorage( + server.api as never, + local, + jest.fn(), + 'default', + tabStore + ); + markClosed(local, 'default', 'newer'); + duplicate.touch(); + + await send(duplicate); + + expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'Y']); + }); + }); }); diff --git a/libs/services/src/lib/server-storage/server-user-data-adopt.ts b/libs/services/src/lib/server-storage/server-user-data-adopt.ts index b596ca092..d8b047c7f 100644 --- a/libs/services/src/lib/server-storage/server-user-data-adopt.ts +++ b/libs/services/src/lib/server-storage/server-user-data-adopt.ts @@ -3,6 +3,7 @@ import { behind, type EditStep, splitEditKey, + uncertain, UNKNOWN_TAG, yielding, } from './server-user-data-entries'; @@ -95,6 +96,17 @@ export function restorePage( page: string ): void { for (const write of readUnsentWrites(local, instanceId, page)) { + const key = `${write.namespace}|${write.playlistId}`; + const queued = writes.isQueued(key) ? writes.unsent(key) : undefined; + if (queued) { + // An older page load's copy came first: these steps follow it. + writes.put({ + ...queued, + value: write.value, + steps: [...queued.steps, ...uncertain(write.steps)], + }); + continue; + } writes.queue( write.namespace, write.playlistId, @@ -107,7 +119,11 @@ export function restorePage( ); } for (const edit of readDeferredEdits(local, instanceId, page)) { - deferred.set(`${edit.namespace}|${edit.playlistId}`, edit); + const key = `${edit.namespace}|${edit.playlistId}`; + deferred.set(key, { + ...edit, + steps: [...(deferred.get(key)?.steps ?? []), ...edit.steps], + }); } } diff --git a/libs/services/src/lib/server-storage/server-user-data-storage.ts b/libs/services/src/lib/server-storage/server-user-data-storage.ts index 99f50c635..4970b20e7 100644 --- a/libs/services/src/lib/server-storage/server-user-data-storage.ts +++ b/libs/services/src/lib/server-storage/server-user-data-storage.ts @@ -177,8 +177,9 @@ export class ServerUserDataStorage implements UserDataStorage { const owner = pages[pages.length - 1]; if (owner && !pageIsLive(this.local, this.instanceId, owner)) { // This page may have read and edited since: adopt, not restore. + // Each adopted page goes ahead of what is queued: newest first. this.awaited = []; - pages.forEach((page) => this.adoptPage(page)); + [...pages].reverse().forEach((page) => this.adoptPage(page)); } // A tab with nothing to sync never re-reads, which is where closed // tabs' saved writes are otherwise picked up.