diff --git a/apps/web-e2e/src/server-storage.e2e.ts b/apps/web-e2e/src/server-storage.e2e.ts index d502d7977..6b08c873f 100644 --- a/apps/web-e2e/src/server-storage.e2e.ts +++ b/apps/web-e2e/src/server-storage.e2e.ts @@ -102,8 +102,13 @@ async function seedBrowserPlaylist( } test.describe('server storage @self-hosted @persistence', () => { - // Both tests share one server database. + // Both tests share one server database, and so would the other browser + // projects: keep it to one browser. test.describe.configure({ mode: 'serial' }); + test.skip( + ({ browserName }) => browserName !== 'chromium', + 'One shared server database' + ); test.beforeEach(async ({ request }) => { const response = await request.delete( diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index 8f832cfb7..972dec327 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -202,7 +202,10 @@ settings and guards read them in every build. (the page-hide flush waits for a batch still travelling), a failed write, including a refused `keepalive` one, stays queued for the next flush, and re-reads run one at a time and keep entries edited while they were in - flight or not yet saved. Start-up waits + flight or not yet saved. Before the first read finishes, an edit is + compared with the cache, so it sends only its own entry. A re-read + updates the cache; views already open show another device's changes on + their next load (no live push). Start-up waits up to 3 s for the first read. - **Connection state.** `ServerStorageStatusService` is `online`, `offline` (a request did not reach the server, or a proxy answered 502, 503 or 504 @@ -227,7 +230,9 @@ settings and guards read them in every build. playlist the server already has, by id or by source (Xtream server and user, Stalker portal and MAC, M3U URL), merges into the server's copy; the others upload with their ids. A merged playlist gains the browser - document's favorites and history, and the legacy Xtream maps follow under + document's favorites and history (M3U channel ids are random per import, + so they are translated through the stream URL; unmatched ones are + dropped), and the legacy Xtream maps follow under the server id; `combineUserData` unions lists (playback positions by content type and Xtream id, renamed to the server playlist) and merges maps, the server's value winning on a conflict. Browser data is never deleted, and diff --git a/libs/services/src/lib/server-storage/server-migration-merge.spec.ts b/libs/services/src/lib/server-storage/server-migration-merge.spec.ts index b2c31f7c9..cca870431 100644 --- a/libs/services/src/lib/server-storage/server-migration-merge.spec.ts +++ b/libs/services/src/lib/server-storage/server-migration-merge.spec.ts @@ -1,5 +1,6 @@ import { combineUserData, + remapChannelReferences, retargetUserData, unionItems, } from './server-migration-merge'; @@ -104,3 +105,32 @@ describe('playback positions', () => { expect(retargetUserData(same, 'b', 'b')).toBe(same); }); }); + +describe('remapChannelReferences', () => { + const browser = [ + { id: 'b1', url: 'http://s/1' }, + { id: 'b2', url: 'http://s/2' }, + { id: 'b3', url: 'http://s/gone' }, + ]; + const server = [ + { id: 's1', url: 'http://s/1' }, + { id: 's2', url: 'http://s/2' }, + ]; + + it('names the server channel with the same stream URL', () => { + expect( + remapChannelReferences(['b2', 'b1', 'b3', 's1'], browser, server) + ).toEqual(['s2', 's1', 's1']); + }); + + it('updates M3U history entries and passes portal items through', () => { + const portal = { id: 7, category_id: 'itv' }; + expect( + remapChannelReferences( + [{ source: 'm3u', id: 'b1', url: 'http://s/1' }, portal], + browser, + server + ) + ).toEqual([{ source: 'm3u', id: 's1', url: 'http://s/1' }, portal]); + }); +}); diff --git a/libs/services/src/lib/server-storage/server-migration-merge.ts b/libs/services/src/lib/server-storage/server-migration-merge.ts index 35bc9da53..fcfd07d9a 100644 --- a/libs/services/src/lib/server-storage/server-migration-merge.ts +++ b/libs/services/src/lib/server-storage/server-migration-merge.ts @@ -71,6 +71,54 @@ function itemKey(item: unknown): string { return `json:${JSON.stringify(item)}`; } +interface ChannelRef { + readonly id?: unknown; + readonly url?: unknown; +} + +/** + * M3U favorites and history name channels by id, and every import gives + * channels new random ids: a browser favorite is translated to the server + * channel with the same stream URL, and dropped when there is none. + * Portal favorites (objects) pass through. + */ +export function remapChannelReferences( + references: readonly T[], + browserChannels: readonly ChannelRef[], + serverChannels: readonly ChannelRef[] +): T[] { + const urlByBrowserId = new Map( + browserChannels.map((channel) => [channel.id, channel.url]) + ); + const serverIds = new Set(serverChannels.map((channel) => channel.id)); + const idByUrl = new Map( + serverChannels.map((channel) => [channel.url, channel.id]) + ); + const result: T[] = []; + for (const reference of references) { + if (typeof reference === 'string') { + const id = serverIds.has(reference) + ? reference + : idByUrl.get(urlByBrowserId.get(reference)); + if (typeof id === 'string') { + result.push(id as T); + } + } else if ( + isPlainObject(reference) && + reference['source'] === 'm3u' && + idByUrl.has(reference['url']) + ) { + result.push({ + ...reference, + id: idByUrl.get(reference['url']), + } as T); + } else { + result.push(reference); + } + } + return result; +} + /** * A browser value moved under another playlist id: list items that name * their playlist (playback positions do) name the server's instead. diff --git a/libs/services/src/lib/server-storage/server-migration.service.spec.ts b/libs/services/src/lib/server-storage/server-migration.service.spec.ts index 4c427a550..131aaa991 100644 --- a/libs/services/src/lib/server-storage/server-migration.service.spec.ts +++ b/libs/services/src/lib/server-storage/server-migration.service.spec.ts @@ -194,7 +194,13 @@ describe('ServerMigrationService', () => { of([ { ...browserPlaylists[0], - favorites: ['http://a', 'http://b'], + favorites: ['b1', 'b2'], + playlist: { + items: [ + { id: 'b1', url: 'http://s/1' }, + { id: 'b2', url: 'http://s/2' }, + ], + }, }, ]), }, @@ -202,12 +208,19 @@ describe('ServerMigrationService', () => { session.documentStore.dbGetAppPlaylistMetas.mockResolvedValue([ { _id: 'server-m', title: 'Family', url: browserPlaylists[0].url }, ]); + // The server imported the same list: same URLs, other channel ids. const stored = { _id: 'server-m', title: 'Family', url: browserPlaylists[0].url, - favorites: ['http://a'], + favorites: ['s1'], recentlyViewed: [], + playlist: { + items: [ + { id: 's1', url: 'http://s/1' }, + { id: 's2', url: 'http://s/2' }, + ], + }, }; session.documentStore.dbGetAppPlaylist.mockResolvedValue(stored); @@ -218,7 +231,7 @@ describe('ServerMigrationService', () => { expect(result).toEqual({ imported: 0, merged: 1, failed: 0 }); expect(session.documentStore.dbUpsertAppPlaylist).toHaveBeenCalledWith({ ...stored, - favorites: ['http://a', 'http://b'], + favorites: ['s1', 's2'], recentlyViewed: [], }); }); diff --git a/libs/services/src/lib/server-storage/server-migration.service.ts b/libs/services/src/lib/server-storage/server-migration.service.ts index 1654d0c2e..61a282500 100644 --- a/libs/services/src/lib/server-storage/server-migration.service.ts +++ b/libs/services/src/lib/server-storage/server-migration.service.ts @@ -9,6 +9,7 @@ import { } from '@iptvnator/shared/interfaces'; import { combineUserData, + remapChannelReferences, retargetUserData, unionItems, } from './server-migration-merge'; @@ -142,12 +143,25 @@ export class ServerMigrationService { if (!stored) { throw new Error('The server playlist disappeared'); } + const browserChannels = channelsOf(playlist); + const serverChannels = channelsOf(stored); const next = { ...stored, - favorites: unionItems(stored.favorites ?? [], favorites), + favorites: unionItems( + stored.favorites ?? [], + remapChannelReferences( + favorites, + browserChannels, + serverChannels + ) + ), recentlyViewed: unionItems( stored.recentlyViewed ?? [], - recentlyViewed + remapChannelReferences( + recentlyViewed, + browserChannels, + serverChannels + ) ), }; if ( @@ -227,6 +241,13 @@ export class ServerMigrationService { } } +/** The M3U channels of a full playlist document. */ +function channelsOf(playlist: Playlist): { id?: unknown; url?: unknown }[] { + const items = (playlist as { playlist?: { items?: unknown } }).playlist + ?.items; + return Array.isArray(items) ? items : []; +} + function readLocal(key: string): string | null { try { return localStorage.getItem(key); diff --git a/libs/services/src/lib/server-storage/server-user-data-storage.spec.ts b/libs/services/src/lib/server-storage/server-user-data-storage.spec.ts index f61b5f563..4035857de 100644 --- a/libs/services/src/lib/server-storage/server-user-data-storage.spec.ts +++ b/libs/services/src/lib/server-storage/server-user-data-storage.spec.ts @@ -324,4 +324,53 @@ describe('ServerUserDataStorage', () => { { value: ['new'] } ); }); + + it('sends only the edited entry before the first read finished', () => { + const { api, local, storage } = setup(); + // The cache from the last visit; no read from the server yet. + local.setItem( + 'iptvnator-server:xtream-favorites', + JSON.stringify({ p1: ['a'], p2: ['b'] }) + ); + + storage.setItem( + 'xtream-favorites', + JSON.stringify({ p1: ['a', 'c'], p2: ['b'] }) + ); + void storage.flush(); + + return settle().then(() => { + expect(api.sendJson.mock.calls).toEqual([ + [ + 'PUT', + '/user-state/xtream-favorites/p1', + { value: ['a', 'c'] }, + ], + ]); + }); + }); + + it('never retries an older page-hide write over a newer one', async () => { + const { api, storage } = setup(); + await storage.hydrate(); + const acks: ((ok: boolean) => void)[] = []; + api.sendKeepalive.mockImplementation( + () => new Promise((resolve) => acks.push(resolve)) + ); + + for (const value of ['A', 'B', 'C']) { + storage.setItem( + 'xtream-favorites', + JSON.stringify({ p1: [value] }) + ); + storage.flushWithKeepalive(); + await settle(); + // Each page-hide write waits for the previous one. + acks[acks.length - 1](value !== 'B'); + await settle(); + } + + await storage.flush(); + expect(api.sendJson).not.toHaveBeenCalled(); + }); }); 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 9e8a0703b..90abfa62d 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 @@ -16,6 +16,8 @@ interface PendingEntry { readonly playlistId: string; /** `undefined` deletes the entry. */ readonly value: unknown; + /** The edit this value came from; a retry never replaces a newer one. */ + readonly edit: number; } type ServerApi = Pick< @@ -64,8 +66,9 @@ export class ServerUserDataStorage implements UserDataStorage { this.local.setItem(key, value); return; } + const before = this.baseline(key); this.local.setItem(SERVER_CACHE_PREFIX + key, value); - this.queueChanges(key, parseMap(value)); + this.queueChanges(key, parseMap(value), before); } removeItem(key: string): void { @@ -73,8 +76,9 @@ export class ServerUserDataStorage implements UserDataStorage { this.local.removeItem(key); return; } + const before = this.baseline(key); this.local.removeItem(SERVER_CACHE_PREFIX + key); - this.queueChanges(key, {}); + this.queueChanges(key, {}, before); } /** @@ -195,31 +199,55 @@ export class ServerUserDataStorage implements UserDataStorage { return isServerUserStateNamespace(key); } + /** + * What a write is compared against: the server's last value, or before + * the first read finishes, the cache, so an edit sends only its entry. + */ + private baseline( + namespace: ServerUserStateNamespace + ): Record { + return ( + this.synced.get(namespace) ?? + parseMap( + this.local.getItem(SERVER_CACHE_PREFIX + namespace) ?? '{}' + ) + ); + } + private queueChanges( namespace: ServerUserStateNamespace, - next: Record + next: Record, + previous: Record ): void { - const previous = this.synced.get(namespace) ?? {}; for (const [playlistId, value] of Object.entries(next)) { if ( JSON.stringify(previous[playlistId]) !== JSON.stringify(value) ) { - this.queue({ namespace, playlistId, value }); + this.queue(namespace, playlistId, value); } } for (const playlistId of Object.keys(previous)) { if (!(playlistId in next)) { - this.queue({ namespace, playlistId, value: undefined }); + this.queue(namespace, playlistId, undefined); } } this.synced.set(namespace, next); this.scheduleFlush(); } - private queue(entry: PendingEntry): void { - const key = entryKey(entry); - this.pending.set(key, entry); - this.edits.set(key, ++this.editCount); + private queue( + namespace: ServerUserStateNamespace, + playlistId: string, + value: unknown + ): void { + const entry: PendingEntry = { + namespace, + playlistId, + value, + edit: ++this.editCount, + }; + this.pending.set(entryKey(entry), entry); + this.edits.set(entryKey(entry), entry.edit); } private scheduleFlush(): void { @@ -246,19 +274,15 @@ export class ServerUserDataStorage implements UserDataStorage { * Back into the queue after a failed write, for the offline retry or * the next edit, unless the entry was edited since it was sent. */ - private requeue( - entry: PendingEntry, - editWhenSent: number | undefined - ): void { + private requeue(entry: PendingEntry): void { const key = entryKey(entry); - if (this.edits.get(key) === editWhenSent && !this.pending.has(key)) { + if (this.edits.get(key) === entry.edit && !this.pending.has(key)) { this.pending.set(key, entry); } } private async sendKeepalive(entry: PendingEntry): Promise { const key = entryKey(entry); - const edit = this.edits.get(key); const path = entryPath(entry); this.keepaliveKeys.add(key); const ok = await (entry.value === undefined @@ -266,12 +290,11 @@ export class ServerUserDataStorage implements UserDataStorage { : this.api.sendKeepalive('PUT', path, { value: entry.value })); this.keepaliveKeys.delete(key); if (!ok) { - this.requeue(entry, edit); + this.requeue(entry); } } private async send(entry: PendingEntry): Promise { - const edit = this.edits.get(entryKey(entry)); try { if (entry.value === undefined) { await this.api.request(entryPath(entry), { method: 'DELETE' }); @@ -281,7 +304,7 @@ export class ServerUserDataStorage implements UserDataStorage { }); } } catch (error) { - this.requeue(entry, edit); + this.requeue(entry); this.onWriteFailure(error); } }