From 676e3adcd47b186c585cbe1a0e36affd5e2325bd Mon Sep 17 00:00:00 2001 From: 4gray Date: Mon, 12 Oct 2026 00:04:56 +0200 Subject: [PATCH] fix(server-storage): keep Xtream data, ordered sync, sign-in and stale-cache fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Moving browser playlists matches them to the server's by id or source and carries the legacy Xtream maps under the server id; it no longer goes through the backup format, whose PWA export has no Xtream user data. A partial move stays on offer. - User-data batches go out in order, and a re-read keeps entries edited while it was in flight. - A keepPayload write keeps its cached copy only when the server replaced the version its payload came from (new X-Replaced-ETag header). - 502/503/504 from the proxy count as offline. - The start-up screen lets the sign-in page through while a token is required. - Settings → Server offers a reload after a move. - The probe detects Electron through RuntimeCapabilitiesService. Co-Authored-By: Claude Opus 5.5 --- .../src/app/storage/playlists-router.ts | 10 +- .../src/app/storage/storage-routes.spec.ts | 11 ++ .../app/app-startup-status.component.spec.ts | 26 ++++ .../src/app/app-startup-status.component.ts | 9 +- .../server-storage/server-instance-probe.ts | 6 +- .../settings-server-section.component.html | 19 ++- .../settings-server-section.component.spec.ts | 1 + .../settings-server-section.component.ts | 8 ++ docs/architecture/pwa-self-hosted.md | 33 ++++-- .../specs/2026-10-08-server-storage-design.md | 15 ++- .../src/lib/playlist-storage-backend.spec.ts | 15 --- .../src/lib/playlist-storage-backend.ts | 20 +--- .../server-storage/server-api-client.spec.ts | 22 ++++ .../lib/server-storage/server-api-client.ts | 8 ++ .../server-migration-plan.spec.ts | 79 +++++++++++++ .../server-storage/server-migration-plan.ts | 94 +++++++++++++++ .../server-migration.service.spec.ts | 111 +++++++++++++----- .../server-migration.service.ts | 111 +++++++++++------- .../server-playlist-document-store.spec.ts | 41 +++++++ .../server-playlist-document-store.ts | 110 ++++++++++------- .../server-storage/server-storage-session.ts | 1 - .../server-user-data-storage.spec.ts | 96 ++++++++++++++- .../server-user-data-storage.ts | 77 +++++++++--- .../src/lib/server-storage.contract.ts | 7 ++ 24 files changed, 732 insertions(+), 198 deletions(-) create mode 100644 libs/services/src/lib/server-storage/server-migration-plan.spec.ts create mode 100644 libs/services/src/lib/server-storage/server-migration-plan.ts diff --git a/apps/web-backend/src/app/storage/playlists-router.ts b/apps/web-backend/src/app/storage/playlists-router.ts index 78a71c553..97d39df99 100644 --- a/apps/web-backend/src/app/storage/playlists-router.ts +++ b/apps/web-backend/src/app/storage/playlists-router.ts @@ -1,4 +1,5 @@ import express, { Router, type Response } from 'express'; +import { SERVER_REPLACED_ETAG_HEADER as REPLACED_ETAG_HEADER } from '@iptvnator/shared/interfaces'; import { matchesEtag } from './etag-match'; import { deleteAllPlaylistsWithState, @@ -53,11 +54,16 @@ export function createPlaylistsRouter( router.put('/playlists/:id', json, (req, res) => { const id = requirePlaylistId(req.params.id); + const replaced = revisions.etag(id); playlists.upsert( { ...requireDocument(req.body), _id: id }, { keepPayload: req.query['keepPayload'] === '1' } ); - withEtag(res, revisions.bump(id)).json({ success: true }); + // With keepPayload a client caches its copy only if the payload it + // holds is the one this write replaced. + withEtag(res, revisions.bump(id)) + .set(REPLACED_ETAG_HEADER, replaced) + .json({ success: true }); }); router.delete('/playlists/:id', (req, res) => { @@ -76,7 +82,7 @@ export function createPlaylistsRouter( function withEtag(res: Response, etag: string): Response { return res .set('ETag', etag) - .set('Access-Control-Expose-Headers', 'ETag') + .set('Access-Control-Expose-Headers', `ETag, ${REPLACED_ETAG_HEADER}`) .set('Cache-Control', 'no-cache'); } diff --git a/apps/web-backend/src/app/storage/storage-routes.spec.ts b/apps/web-backend/src/app/storage/storage-routes.spec.ts index 88a3ac911..d96c1c65f 100644 --- a/apps/web-backend/src/app/storage/storage-routes.spec.ts +++ b/apps/web-backend/src/app/storage/storage-routes.spec.ts @@ -91,6 +91,17 @@ describe('server storage routes', () => { headers: { 'if-none-match': `W/${etag}` }, }); expect(weakened.status).toBe(304); + + // A write names the version it replaced. + const next = await fetch( + `${baseUrl}/v1/playlists/p1?keepPayload=1`, + json({ _id: 'p1', title: 'Renamed' }, { method: 'PUT' }) + ); + expect(next.headers.get('x-replaced-etag')).toBe(etag); + expect(next.headers.get('etag')).not.toBe(etag); + expect( + next.headers.get('access-control-expose-headers') + ).toContain('X-Replaced-ETag'); }) )); diff --git a/apps/web/src/app/app-startup-status.component.spec.ts b/apps/web/src/app/app-startup-status.component.spec.ts index dae454396..8e94f7b18 100644 --- a/apps/web/src/app/app-startup-status.component.spec.ts +++ b/apps/web/src/app/app-startup-status.component.spec.ts @@ -6,6 +6,7 @@ import { selectPlaylistsLoadFailed, selectPlaylistsLoadingFlag, } from '@iptvnator/m3u-state'; +import { ServerStorageStatusService } from '@iptvnator/services'; import { AppStartupStatusComponent } from './app-startup-status.component'; describe('AppStartupStatusComponent', () => { @@ -48,6 +49,31 @@ describe('AppStartupStatusComponent', () => { expect(fixture.componentInstance.complete()).toBe(true); }); + it('shows the server sign-in page although sources could not load', () => { + TestBed.inject(MockStore).overrideSelector( + selectPlaylistsLoadFailed, + true + ); + const status = TestBed.inject(ServerStorageStatusService); + status.activate( + { + service: 'iptvnator', + apiVersion: 1, + instanceId: 'i', + storage: 'server', + auth: 'token', + }, + () => undefined + ); + status.reportUnauthorized(); + const fixture = TestBed.createComponent(AppStartupStatusComponent); + fixture.componentRef.setInput('routeReady', true); + fixture.detectChanges(); + + expect(fixture.componentInstance.complete()).toBe(true); + expect(fixture.nativeElement.querySelector('section')).toBeNull(); + }); + it('offers a retry on failure without rendering raw storage errors', () => { TestBed.inject(MockStore).overrideSelector( selectPlaylistsLoadFailed, diff --git a/apps/web/src/app/app-startup-status.component.ts b/apps/web/src/app/app-startup-status.component.ts index 46375638c..ba421aa4e 100644 --- a/apps/web/src/app/app-startup-status.component.ts +++ b/apps/web/src/app/app-startup-status.component.ts @@ -14,6 +14,7 @@ import { selectPlaylistsLoadFailed, selectPlaylistsLoadingFlag, } from '@iptvnator/m3u-state'; +import { ServerStorageStatusService } from '@iptvnator/services'; /** The first route also waits for settings and XMLTV source reconciliation. */ @Component({ @@ -130,8 +131,14 @@ export class AppStartupStatusComponent { selectPlaylistsLoadingFlag ); readonly failed = this.store.selectSignal(selectPlaylistsLoadFailed); + private readonly serverStatus = inject(ServerStorageStatusService); + /** Server storage waiting for a token: only its sign-in page can load. */ + private readonly signInRequired = computed( + () => this.serverStatus.state() === 'unauthorized' + ); readonly complete = computed( - () => this.routeReady() && this.sourcesReady() + () => + this.routeReady() && (this.sourcesReady() || this.signInRequired()) ); retry(): void { diff --git a/apps/web/src/app/server-storage/server-instance-probe.ts b/apps/web/src/app/server-storage/server-instance-probe.ts index 20e6e9bc6..4d7d9766c 100644 --- a/apps/web/src/app/server-storage/server-instance-probe.ts +++ b/apps/web/src/app/server-storage/server-instance-probe.ts @@ -1,3 +1,4 @@ +import { RuntimeCapabilitiesService } from '@iptvnator/services'; import type { ServerInstanceInfo } from '@iptvnator/shared/interfaces'; import type { IptvnatorRuntimeConfig } from '../services/runtime-config'; @@ -29,7 +30,10 @@ export async function probeServerInstance( ): Promise { const runtimeConfig = options.runtimeConfig ?? globalThis.window?.__IPTVNATOR_CONFIG__; - const isElectron = options.isElectron ?? !!globalThis.window?.electron; + // The capability service has no dependencies, so it works before + // bootstrap and keeps one definition of the desktop runtime. + const isElectron = + options.isElectron ?? new RuntimeCapabilitiesService().isElectron; const baseUrl = runtimeConfig?.BACKEND_URL?.trim(); if (isElectron || !baseUrl) { return { kind: 'skipped' }; diff --git a/apps/web/src/app/settings/settings-server-section.component.html b/apps/web/src/app/settings/settings-server-section.component.html index f080a71ff..641379835 100644 --- a/apps/web/src/app/settings/settings-server-section.component.html +++ b/apps/web/src/app/settings/settings-server-section.component.html @@ -91,8 +91,19 @@ }

- @if (browserPlaylistCount() > 0) { -
+
+ @if (movedSomething()) { + + + } + @if (browserPlaylistCount() > 0) {
- } + } +
} diff --git a/apps/web/src/app/settings/settings-server-section.component.spec.ts b/apps/web/src/app/settings/settings-server-section.component.spec.ts index 9fdfd8d6a..04ff8cf8b 100644 --- a/apps/web/src/app/settings/settings-server-section.component.spec.ts +++ b/apps/web/src/app/settings/settings-server-section.component.spec.ts @@ -92,6 +92,7 @@ describe('SettingsServerSectionComponent', () => { expect(migration.moveBrowserData).toHaveBeenCalled(); expect(byTestId('server-import-browser')).toBeNull(); + expect(byTestId('server-import-reload')).not.toBeNull(); expect(fixture.nativeElement.textContent).toContain( 'SERVER_STORAGE.MIGRATION_DONE' ); diff --git a/apps/web/src/app/settings/settings-server-section.component.ts b/apps/web/src/app/settings/settings-server-section.component.ts index 5e2c54173..7da1af19c 100644 --- a/apps/web/src/app/settings/settings-server-section.component.ts +++ b/apps/web/src/app/settings/settings-server-section.component.ts @@ -61,6 +61,10 @@ export class SettingsServerSectionComponent { readonly showBrowserImport = computed( () => this.browserPlaylistCount() > 0 || this.importResult() !== null ); + readonly movedSomething = computed(() => { + const result = this.importResult(); + return !!result && result.imported + result.merged > 0; + }); constructor() { void this.load(); @@ -99,6 +103,10 @@ export class SettingsServerSectionComponent { signOut(): void { this.session.signOut(); } + + reload(): void { + window.location.reload(); + } } function backendOrigin(baseUrl: string): string { diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index e2482857b..329ca7552 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -83,7 +83,9 @@ Server storage adds versioned routes under `/v1`, which the PWA uses in `DELETE /v1/playlists`: whole playlist documents; `GET` sends an `ETag` and answers `304` to a matching `If-None-Match`, and `PUT ...?keepPayload=1` keeps the stored `playlist`, `items` and - `header` fields (`409 payload_missing` without a stored row) + `header` fields (`409 payload_missing` without a stored row); every `PUT` + answers with the new `ETag` and `X-Replaced-ETag`, the version it + replaced - `GET /v1/user-state`, `PUT|DELETE /v1/user-state/:namespace/:playlistId`: Xtream favorites, recently viewed, collection snapshots and playback positions, one JSON value per namespace and playlist @@ -187,34 +189,39 @@ settings and guards read them in every build. `electron`, `server`, `indexeddb`), so `PlaylistsService` is unchanged. The store caches full documents by ETag and uploads a document whose payload it served unchanged with `keepPayload=1`, so a favorite or a rename does not - resend a large M3U. + resend a large M3U. It keeps its copy after such a write only when + `X-Replaced-ETag` names the version its payload came from; otherwise + another device wrote in between and the next read fetches the document. - **Xtream user data.** `USER_DATA_STORAGE` becomes `ServerUserDataStorage`: the four synced namespaces are cached under the `iptvnator-server:` key prefix (the legacy unprefixed keys are never written in server mode) and each change sends only the playlist entries that differ, debounced, with a - `keepalive` flush when the page hides. Start-up waits up to 3 s for the - first read. + `keepalive` flush when the page hides. Batches go out one after another, + and a re-read keeps entries edited while it was in flight. Start-up waits + up to 3 s for the first read. - **Connection state.** `ServerStorageStatusService` is `online`, `offline` - (a request did not reach the server; retried every 30 s and when the tab - returns) or + (a request did not reach the server, or a proxy answered 502, 503 or 504 + for it; retried every 30 s and when the tab returns) or `unauthorized` (a 401 clears the stored token). The rail shows a status shortcut, an offline banner sits above the workspace content, and failed writes raise a snackbar (`SERVER_STORAGE.WRITE_FAILED`, or `TOO_LARGE` for a 413). - **Access token.** With `IPTVNATOR_ACCESS_TOKEN` the instance reports `auth: 'token'`; without a stored token, or after a 401, the guards route to - `/server-sign-in`, which checks the token against `/v1/instance/status` + `/server-sign-in` (the start-up screen lets it through although the + playlists could not load), which checks the token against `/v1/instance/status` before it stores it under `iptvnator-server:access-token` and reloads. `ServerAccessTokenInterceptor` adds the header to `HttpClient` calls to the backend, and `PwaService` adds it to its two raw `fetch` calls. - **Moving browser data.** `ServerMigrationService` offers, once per server instance (marker `iptvnator-server:migration:`), to move the - playlists this browser kept in IndexedDB. An empty server receives them - with their ids plus the legacy Xtream maps for those ids; a server that - already has playlists goes through the backup export and import, whose - fingerprints merge duplicates. Browser data is never deleted. Settings → - Server repeats the offer while the browser has playlists that were not - moved. + playlists this browser kept in IndexedDB (`planServerMigration`). A + 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. The legacy Xtream maps follow under the + server id without replacing what a merged playlist already has. Browser + data is never deleted, and the marker is written only when nothing + failed, so Settings → Server keeps offering a partial move. What stays per device: settings, parental lock PINs and locked groups, the global favorites order and other UI preferences in browser storage. diff --git a/docs/superpowers/specs/2026-10-08-server-storage-design.md b/docs/superpowers/specs/2026-10-08-server-storage-design.md index 4a5619f49..851ff48f2 100644 --- a/docs/superpowers/specs/2026-10-08-server-storage-design.md +++ b/docs/superpowers/specs/2026-10-08-server-storage-design.md @@ -79,9 +79,15 @@ on the server. Adding a playlist stays the ordinary "Add playlist" flow. this instance): a dialog offers "Move to server". Playlists are copied with their ids, then the four Xtream user-data maps for those ids. - Any later device (server already has playlists): Settings → Server - offers "Import this browser's playlists", which runs the existing backup - export (browser) and import (server) with fingerprint merging, so the - same provider added on two devices is not duplicated. + offers "Import this browser's playlists". A browser playlist the server + already has (same id, or same source: Xtream server and user, Stalker + portal and MAC, M3U URL) merges into the server's copy, so the same + provider added on two devices is not duplicated; the others upload with + their ids. The Xtream maps follow under the server id and never replace + what a merged playlist already has. The backup format is not used: its + PWA export carries no Xtream user data. + - The decision marker is written only when nothing failed, so a partial + move stays on offer and can run again. - Browser data is never deleted. The server-mode cache uses its own localStorage key prefix, so legacy keys stay intact. 5. Offline: reads and writes fail with a visible state. The rail shows an @@ -143,8 +149,7 @@ closes it on `SIGTERM`. Storage errors are logged by name and route only. - `PlaylistStorageBackend` (libs/services) resolves the document store: server store in server mode, Electron bridge when `supportsSqlite`, else - IndexedDB. It can switch to browser storage temporarily for the migration - export. + IndexedDB. - `@iptvnator/services/server-storage` (sub-entry, kept out of the eager barrel): API client, document store, user-data storage, session, migration. - `ServerStorageStatusService` (main barrel, tiny): connection state for the diff --git a/libs/services/src/lib/playlist-storage-backend.spec.ts b/libs/services/src/lib/playlist-storage-backend.spec.ts index 50c50bd75..57da94822 100644 --- a/libs/services/src/lib/playlist-storage-backend.spec.ts +++ b/libs/services/src/lib/playlist-storage-backend.spec.ts @@ -52,19 +52,4 @@ describe('PlaylistStorageBackend', () => { expect(backend.kind()).toBe('server'); expect(backend.documentStore()).toBe(store); }); - - it('falls back to browser storage while an override task runs', async () => { - testWindow.electron = undefined; - const backend = create(false); - backend.useServerStore({ dbGetAppPlaylists: jest.fn() } as never); - - const seen = await backend.runWithBrowserStorage(async () => [ - backend.kind(), - backend.usesBrowserStorageOverride(), - ]); - - expect(seen).toEqual(['indexeddb', true]); - expect(backend.kind()).toBe('server'); - expect(backend.usesBrowserStorageOverride()).toBe(false); - }); }); diff --git a/libs/services/src/lib/playlist-storage-backend.ts b/libs/services/src/lib/playlist-storage-backend.ts index 983c32092..b6ab6a478 100644 --- a/libs/services/src/lib/playlist-storage-backend.ts +++ b/libs/services/src/lib/playlist-storage-backend.ts @@ -35,32 +35,14 @@ export type PlaylistStorageKind = 'electron' | 'server' | 'indexeddb'; export class PlaylistStorageBackend { private readonly runtime = inject(RuntimeCapabilitiesService); private serverStore: PlaylistDocumentStore | null = null; - private browserOverrideDepth = 0; /** Server mode: set once by the server storage session at startup. */ useServerStore(store: PlaylistDocumentStore): void { this.serverStore = store; } - usesBrowserStorageOverride(): boolean { - return this.browserOverrideDepth > 0; - } - - /** - * Runs `task` against browser storage although server mode is active. - * Only the migration export uses it, inside a modal flow. - */ - async runWithBrowserStorage(task: () => Promise): Promise { - this.browserOverrideDepth += 1; - try { - return await task(); - } finally { - this.browserOverrideDepth -= 1; - } - } - kind(): PlaylistStorageKind { - if (this.serverStore && this.browserOverrideDepth === 0) { + if (this.serverStore) { return 'server'; } return this.runtime.supportsSqlite ? 'electron' : 'indexeddb'; diff --git a/libs/services/src/lib/server-storage/server-api-client.spec.ts b/libs/services/src/lib/server-storage/server-api-client.spec.ts index a61c6ad8e..0f9843631 100644 --- a/libs/services/src/lib/server-storage/server-api-client.spec.ts +++ b/libs/services/src/lib/server-storage/server-api-client.spec.ts @@ -62,6 +62,28 @@ describe('ServerApiClient', () => { expect(onWriteFailure).toHaveBeenCalledTimes(1); }); + it('reports offline when a proxy answers for a down backend', async () => { + for (const code of [502, 503, 504]) { + const fetchFn = jest + .fn() + .mockResolvedValue(response(code, 'Bad Gateway')); + + await expect( + client(fetchFn).getJson('/user-state') + ).rejects.toMatchObject({ status: code }); + } + expect(status.reportOffline).toHaveBeenCalledTimes(3); + + // A backend error is not an outage. + status.reportOffline.mockClear(); + await expect( + client( + jest.fn().mockResolvedValue(response(500, { code: 'x' })) + ).getJson('/user-state') + ).rejects.toMatchObject({ status: 500 }); + expect(status.reportOffline).not.toHaveBeenCalled(); + }); + it('clears the token and reports unauthorized on 401', async () => { localStorage.setItem(SERVER_ACCESS_TOKEN_STORAGE_KEY, 'old'); const fetchFn = jest diff --git a/libs/services/src/lib/server-storage/server-api-client.ts b/libs/services/src/lib/server-storage/server-api-client.ts index 92d36a91b..71b89645f 100644 --- a/libs/services/src/lib/server-storage/server-api-client.ts +++ b/libs/services/src/lib/server-storage/server-api-client.ts @@ -7,6 +7,9 @@ import type { ServerStorageStatusService } from '../server-storage-status.servic const REQUEST_TIMEOUT_MS = 20_000; +/** nginx answers these while the backend behind it is down. */ +const GATEWAY_UNAVAILABLE = new Set([502, 503, 504]); + export class ServerApiError extends Error { constructor( readonly status: number, @@ -78,6 +81,11 @@ export class ServerApiClient { ); } + if (GATEWAY_UNAVAILABLE.has(response.status)) { + // A proxy in front of a backend that is down or restarting. + this.options.status.reportOffline(); + } + if (!response.ok && response.status !== 304) { const body = await readErrorBody(response); const error = new ServerApiError( diff --git a/libs/services/src/lib/server-storage/server-migration-plan.spec.ts b/libs/services/src/lib/server-storage/server-migration-plan.spec.ts new file mode 100644 index 000000000..5ad8fbe35 --- /dev/null +++ b/libs/services/src/lib/server-storage/server-migration-plan.spec.ts @@ -0,0 +1,79 @@ +import type { Playlist } from '@iptvnator/shared/interfaces'; +import { planServerMigration, sourceIdentity } from './server-migration-plan'; + +const playlist = (fields: Partial & { _id: string }) => + ({ title: fields._id, ...fields }) as Playlist; + +describe('planServerMigration', () => { + it('uploads everything to an empty server', () => { + const browser = [ + playlist({ _id: 'm', url: 'http://a.test/list.m3u' }), + playlist({ _id: 'file' }), + ]; + + expect(planServerMigration(browser, [])).toEqual([ + { kind: 'upload', playlist: browser[0] }, + { kind: 'upload', playlist: browser[1] }, + ]); + }); + + it('merges by id first, then by source', () => { + const server = [ + playlist({ _id: 'moved' }), + playlist({ + _id: 's-x', + serverUrl: 'HTTP://Panel.test/', + username: 'u', + }), + playlist({ + _id: 's-st', + portalUrl: 'http://portal.test/c', + macAddress: '00:1A:79:AA:BB:CC', + }), + playlist({ _id: 's-m', url: 'http://a.test/list.m3u' }), + ]; + const browser = [ + playlist({ _id: 'moved' }), + playlist({ + _id: 'b-x', + serverUrl: 'http://panel.test', + username: 'u', + }), + playlist({ + _id: 'b-st', + url: 'http://portal.test/c/', + macAddress: '00:1a:79:aa:bb:cc', + }), + playlist({ _id: 'b-m', url: 'http://a.test/list.m3u' }), + playlist({ + _id: 'b-other', + serverUrl: 'http://panel.test', + username: 'other', + }), + ]; + + expect(planServerMigration(browser, server)).toEqual([ + { kind: 'merge', browserId: 'moved', serverId: 'moved' }, + { kind: 'merge', browserId: 'b-x', serverId: 's-x' }, + { kind: 'merge', browserId: 'b-st', serverId: 's-st' }, + { kind: 'merge', browserId: 'b-m', serverId: 's-m' }, + { kind: 'upload', playlist: browser[4] }, + ]); + }); + + it('collapses browser duplicates of one source onto the first', () => { + const browser = [ + playlist({ _id: 'one', url: 'http://a.test/list.m3u' }), + playlist({ _id: 'two', url: 'http://A.test/list.m3u' }), + ]; + + expect(planServerMigration(browser, [])).toEqual([ + { kind: 'upload', playlist: browser[0] }, + { kind: 'merge', browserId: 'two', serverId: 'one' }, + ]); + }); + + it('has no source for a file or pasted M3U', () => { + expect(sourceIdentity(playlist({ _id: 'file' }))).toBeNull(); + }); +}); diff --git a/libs/services/src/lib/server-storage/server-migration-plan.ts b/libs/services/src/lib/server-storage/server-migration-plan.ts new file mode 100644 index 000000000..f353bc79a --- /dev/null +++ b/libs/services/src/lib/server-storage/server-migration-plan.ts @@ -0,0 +1,94 @@ +import type { Playlist } from '@iptvnator/shared/interfaces'; + +/** Where one browser playlist goes on the server. */ +export type ServerMigrationStep = + | { + readonly kind: 'upload'; + readonly playlist: Playlist; + } + | { + /** The server already has this source; it keeps its document. */ + readonly kind: 'merge'; + readonly browserId: string; + readonly serverId: string; + }; + +/** + * Matches browser playlists to the server's: the same id first (a playlist + * moved before), then the same source (Xtream server and user, Stalker + * portal and MAC, M3U URL). A file or pasted M3U has no source to compare + * and is uploaded unless its id is taken. Browser duplicates of one source + * collapse onto the first. + */ +export function planServerMigration( + browser: readonly Playlist[], + server: readonly Playlist[] +): ServerMigrationStep[] { + const serverIds = new Set(server.map((playlist) => playlist._id)); + const bySource = new Map(); + for (const playlist of server) { + const source = sourceIdentity(playlist); + if (source && !bySource.has(source)) { + bySource.set(source, playlist._id); + } + } + + const steps: ServerMigrationStep[] = []; + for (const playlist of browser) { + const source = sourceIdentity(playlist); + const match = serverIds.has(playlist._id) + ? playlist._id + : source + ? bySource.get(source) + : undefined; + if (match) { + steps.push({ + kind: 'merge', + browserId: playlist._id, + serverId: match, + }); + continue; + } + steps.push({ kind: 'upload', playlist }); + serverIds.add(playlist._id); + if (source) { + bySource.set(source, playlist._id); + } + } + return steps; +} + +/** The provider account or URL a playlist reads from, if it has one. */ +export function sourceIdentity(playlist: Playlist): string | null { + if (playlist.serverUrl && playlist.username) { + return [ + 'xtream', + normalizeUrl(playlist.serverUrl), + playlist.username.trim(), + ].join('|'); + } + const portalUrl = playlist.portalUrl ?? playlist.url; + if (playlist.macAddress && portalUrl) { + return [ + 'stalker', + normalizeUrl(portalUrl), + playlist.macAddress.trim().toLowerCase(), + ].join('|'); + } + return playlist.url ? `m3u|${normalizeUrl(playlist.url)}` : null; +} + +/** Same rules as the backup import's fingerprints. */ +function normalizeUrl(value: string): string { + const trimmed = value.trim(); + try { + const url = new URL(trimmed); + const pathname = + url.pathname === '/' + ? '/' + : url.pathname.replace(/\/+$/, '') || '/'; + return `${url.protocol.toLowerCase()}//${url.host.toLowerCase()}${pathname}${url.search}${url.hash}`; + } catch { + return trimmed.toLowerCase(); + } +} 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 4f52b042d..cda559700 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 @@ -1,46 +1,45 @@ import { TestBed } from '@angular/core/testing'; import { NgxIndexedDBService } from 'ngx-indexed-db'; import { of } from 'rxjs'; -import { PlaylistStorageBackend } from '../playlist-storage-backend'; -import { PlaylistBackupService } from '../playlist-backup.service'; import { SERVER_STORAGE_CONFIG } from './server-storage.config'; import { ServerStorageSession } from './server-storage-session'; import { ServerMigrationService } from './server-migration.service'; describe('ServerMigrationService', () => { const browserPlaylists = [ - { _id: 'm', title: 'Family' }, - { _id: 'x', title: 'Provider', serverUrl: 'http://panel.test' }, + { _id: 'm', title: 'Family', url: 'http://lists.test/family.m3u' }, + { + _id: 'x', + title: 'Provider', + serverUrl: 'http://panel.test', + username: 'user', + }, ]; let session: { fetchStatus: jest.Mock; - documentStore: { dbUpsertAppPlaylist: jest.Mock }; - api: { sendJson: jest.Mock }; + documentStore: { + dbGetAppPlaylistMetas: jest.Mock; + dbUpsertAppPlaylist: jest.Mock; + }; + api: { getJson: jest.Mock; sendJson: jest.Mock }; userData: { hydrate: jest.Mock }; }; - let backup: { exportBackup: jest.Mock; importBackup: jest.Mock }; beforeEach(() => { session = { fetchStatus: jest.fn().mockResolvedValue({ playlists: 0 }), documentStore: { + dbGetAppPlaylistMetas: jest.fn().mockResolvedValue([]), dbUpsertAppPlaylist: jest .fn() .mockResolvedValue({ success: true }), }, - api: { sendJson: jest.fn().mockResolvedValue(undefined) }, + api: { + getJson: jest.fn().mockResolvedValue({ entries: {} }), + sendJson: jest.fn().mockResolvedValue(undefined), + }, userData: { hydrate: jest.fn().mockResolvedValue(undefined) }, }; - backup = { - exportBackup: jest.fn().mockResolvedValue({ json: '{}' }), - importBackup: jest.fn().mockResolvedValue({ - imported: 1, - merged: 1, - skipped: 0, - failed: 0, - errors: [], - }), - }; TestBed.configureTestingModule({ providers: [ { @@ -48,14 +47,6 @@ describe('ServerMigrationService', () => { useValue: { getAll: () => of(browserPlaylists) }, }, { provide: ServerStorageSession, useValue: session }, - { provide: PlaylistBackupService, useValue: backup }, - { - provide: PlaylistStorageBackend, - useValue: { - runWithBrowserStorage: (task: () => Promise) => - task(), - }, - }, { provide: SERVER_STORAGE_CONFIG, useValue: { @@ -111,18 +102,76 @@ describe('ServerMigrationService', () => { { value: ['live:1'] } ); expect(session.api.sendJson).toHaveBeenCalledTimes(1); + expect(session.userData.hydrate).toHaveBeenCalled(); expect(migration.browserDataMoved()).toBe(true); }); - it('merges through the backup format when the server already has playlists', async () => { - session.fetchStatus.mockResolvedValue({ playlists: 4 }); + it('merges into the server copy and moves Xtream data under its id', async () => { + session.documentStore.dbGetAppPlaylistMetas.mockResolvedValue([ + { + _id: 'server-x', + title: 'Provider', + serverUrl: 'http://panel.test/', + username: 'user', + }, + { + _id: 'server-m', + title: 'Family', + url: 'http://lists.test/family.m3u', + }, + ]); + session.api.getJson.mockResolvedValue({ + entries: { + 'xtream-recent-items': { 'server-x': ['vod:9'] }, + }, + }); + localStorage.setItem( + 'xtream-favorites', + JSON.stringify({ x: ['live:1'] }) + ); + localStorage.setItem( + 'xtream-recent-items', + JSON.stringify({ x: ['vod:1'] }) + ); const result = await TestBed.inject( ServerMigrationService ).moveBrowserData(); - expect(backup.exportBackup).toHaveBeenCalled(); - expect(backup.importBackup).toHaveBeenCalledWith('{}'); - expect(result).toEqual({ imported: 1, merged: 1, failed: 0 }); + expect(result).toEqual({ imported: 0, merged: 2, failed: 0 }); + expect( + session.documentStore.dbUpsertAppPlaylist + ).not.toHaveBeenCalled(); + // Favorites the server copy lacks follow; its own history stays. + expect(session.api.sendJson.mock.calls).toEqual([ + [ + 'PUT', + '/user-state/xtream-favorites/server-x', + { value: ['live:1'] }, + ], + ]); + }); + + it('keeps the move on offer when a playlist could not be uploaded', async () => { + session.documentStore.dbUpsertAppPlaylist + .mockRejectedValueOnce(new Error('413')) + .mockResolvedValue({ success: true }); + const migration = TestBed.inject(ServerMigrationService); + + const result = await migration.moveBrowserData(); + + expect(result).toEqual({ imported: 1, merged: 0, failed: 1 }); + expect(migration.browserDataMoved()).toBe(false); + + // A second run merges the moved one and retries the other. + session.documentStore.dbGetAppPlaylistMetas.mockResolvedValue([ + browserPlaylists[1], + ]); + await expect(migration.moveBrowserData()).resolves.toEqual({ + imported: 1, + merged: 1, + failed: 0, + }); + expect(migration.browserDataMoved()).toBe(true); }); }); 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 08dbe10bc..1f5df5461 100644 --- a/libs/services/src/lib/server-storage/server-migration.service.ts +++ b/libs/services/src/lib/server-storage/server-migration.service.ts @@ -5,9 +5,9 @@ import { DbStores, SERVER_USER_STATE_NAMESPACES, type Playlist, + type ServerUserStateResponse, } from '@iptvnator/shared/interfaces'; -import { PlaylistBackupService } from '../playlist-backup.service'; -import { PlaylistStorageBackend } from '../playlist-storage-backend'; +import { planServerMigration } from './server-migration-plan'; import { SERVER_STORAGE_CONFIG } from './server-storage.config'; import { ServerStorageSession } from './server-storage-session'; @@ -25,18 +25,24 @@ export interface ServerMigrationResult { readonly failed: number; } +/** A browser playlist's id on the server, and whether it arrived just now. */ +interface MigrationTarget { + readonly serverId: string; + readonly uploaded: boolean; +} + /** - * Moves this browser's playlists to the server. An empty server receives - * them with their ids plus the legacy Xtream user-data maps; a server that - * already has playlists goes through the backup export/import, whose - * fingerprints merge duplicates. Browser data is never deleted. + * Moves this browser's playlists to the server. New playlists keep their + * ids; one the server already has (same id, or same provider source) is + * merged into the server's copy. The legacy Xtream favorites, history, + * collections and playback positions follow under the server id, without + * replacing what a merged playlist already has there. Browser data is + * never deleted, and the move can run again after a failure. */ @Injectable({ providedIn: 'root' }) export class ServerMigrationService { private readonly dbService = inject(NgxIndexedDBService); private readonly session = inject(ServerStorageSession); - private readonly backend = inject(PlaylistStorageBackend); - private readonly backup = inject(PlaylistBackupService); private readonly instanceId = inject(SERVER_STORAGE_CONFIG).instance .instanceId; @@ -75,59 +81,82 @@ export class ServerMigrationService { } async moveBrowserData(): Promise { - const { playlists: serverCount } = await this.session.fetchStatus(); - const result = - serverCount === 0 - ? await this.copyWithIds() - : await this.mergeThroughBackup(); - this.writeMarker('done'); - return result; - } - - private async copyWithIds(): Promise { - const playlists = await this.readBrowserPlaylists(); + const [browser, server] = await Promise.all([ + this.readBrowserPlaylists(), + this.session.documentStore.dbGetAppPlaylistMetas(), + ]); + const targets = new Map(); let imported = 0; + let merged = 0; let failed = 0; - for (const playlist of playlists) { + + for (const step of planServerMigration(browser, server)) { + if (step.kind === 'merge') { + merged += 1; + targets.set(step.browserId, { + serverId: step.serverId, + uploaded: false, + }); + continue; + } try { - await this.session.documentStore.dbUpsertAppPlaylist(playlist); + await this.session.documentStore.dbUpsertAppPlaylist( + step.playlist + ); imported += 1; + targets.set(step.playlist._id, { + serverId: step.playlist._id, + uploaded: true, + }); } catch { failed += 1; } } - await this.copyXtreamUserData(new Set(playlists.map((p) => p._id))); - return { imported, merged: 0, failed }; + + failed += await this.copyXtreamUserData(targets); + await this.session.userData.hydrate(); + if (failed === 0) { + // A partial move stays on offer, in Settings → Server. + this.writeMarker('done'); + } + return { imported, merged, failed }; } + /** Copies the legacy maps; returns how many entries failed. */ private async copyXtreamUserData( - playlistIds: ReadonlySet - ): Promise { + targets: ReadonlyMap + ): Promise { + if (targets.size === 0) { + return 0; + } + const { entries } = + await this.session.api.getJson( + '/user-state' + ); + let failed = 0; for (const namespace of SERVER_USER_STATE_NAMESPACES) { const legacy = parseMap(readLocal(namespace)); - for (const [playlistId, value] of Object.entries(legacy)) { - if (playlistIds.has(playlistId)) { + const onServer = entries[namespace] ?? {}; + for (const [browserId, value] of Object.entries(legacy)) { + const target = targets.get(browserId); + if ( + !target || + (!target.uploaded && target.serverId in onServer) + ) { + continue; + } + try { await this.session.api.sendJson( 'PUT', - `/user-state/${namespace}/${encodeURIComponent(playlistId)}`, + `/user-state/${namespace}/${encodeURIComponent(target.serverId)}`, { value } ); + } catch { + failed += 1; } } } - await this.session.userData.hydrate(); - } - - private async mergeThroughBackup(): Promise { - const { json } = await this.backend.runWithBrowserStorage(() => - this.backup.exportBackup() - ); - const summary = await this.backup.importBackup(json); - return { - imported: summary.imported, - merged: summary.merged, - failed: summary.failed, - }; + return failed; } private readBrowserPlaylists(): Promise { diff --git a/libs/services/src/lib/server-storage/server-playlist-document-store.spec.ts b/libs/services/src/lib/server-storage/server-playlist-document-store.spec.ts index a66a69b52..0939b4ff2 100644 --- a/libs/services/src/lib/server-storage/server-playlist-document-store.spec.ts +++ b/libs/services/src/lib/server-storage/server-playlist-document-store.spec.ts @@ -11,6 +11,15 @@ function fakeResponse(status: number, body?: unknown, etag?: string) { } as unknown as Response; } +function putResponse(etag: string, replaced: string) { + return { + status: 200, + ok: true, + headers: new Headers({ etag, 'x-replaced-etag': replaced }), + json: async () => ({ success: true }), + } as unknown as Response; +} + describe('ServerPlaylistDocumentStore', () => { const doc = { _id: 'p1', @@ -99,4 +108,36 @@ describe('ServerPlaylistDocumentStore', () => { expect(api.sendJson.mock.calls[1][1]).toBe('/playlists/p1'); expect(api.sendJson.mock.calls[1][2]).toBe(served); }); + + describe('after a keepPayload write', () => { + async function readThenEdit(replaced: string) { + const { api, store } = setup(); + // nginx gzip weakens the read's ETag. + api.request.mockResolvedValueOnce( + fakeResponse(200, doc, 'W/"b.0.1"') + ); + const served = await store.dbGetAppPlaylist('p1'); + if (!served) throw new Error('expected a document'); + api.sendJson.mockResolvedValueOnce( + putResponse('"b.0.2"', replaced) + ); + await store.dbUpsertAppPlaylist({ ...served, favorites: ['a'] }); + api.request.mockResolvedValueOnce( + fakeResponse(200, doc, '"b.0.3"') + ); + await store.dbGetAppPlaylist('p1'); + return api.request.mock.calls[1][1]; + } + + it('keeps its copy when it replaced the version it read', async () => { + await expect(readThenEdit('"b.0.1"')).resolves.toEqual({ + headers: { 'If-None-Match': '"b.0.2"' }, + }); + }); + + it('drops its copy when another device wrote in between', async () => { + // The server kept a newer payload than the one this client has. + await expect(readThenEdit('"b.0.4"')).resolves.toEqual({}); + }); + }); }); diff --git a/libs/services/src/lib/server-storage/server-playlist-document-store.ts b/libs/services/src/lib/server-storage/server-playlist-document-store.ts index 9b926929b..037d7d892 100644 --- a/libs/services/src/lib/server-storage/server-playlist-document-store.ts +++ b/libs/services/src/lib/server-storage/server-playlist-document-store.ts @@ -1,5 +1,6 @@ import { SERVER_PAYLOAD_FIELDS, + SERVER_REPLACED_ETAG_HEADER, type ElectronBridgeCountResult, type ElectronBridgeResult, type Playlist, @@ -16,14 +17,15 @@ type ServerApi = Pick; /** * `PlaylistDocumentStore` over the server storage API. Full documents are - * cached by ETag; every read hands out a private copy and remembers its - * payload objects, so an upsert whose payload is one of them omits it - * (`keepPayload=1`). `PlaylistsService` builds a new object per mutation, - * so cached documents are never mutated in place. + * cached by ETag; every read hands out a private copy and remembers which + * ETag its payload objects came from, so an upsert whose payload is one of + * them omits it (`keepPayload=1`). `PlaylistsService` builds a new object + * per mutation, so cached documents are never mutated in place. */ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { private readonly cache = new Map(); - private readonly servedPayloads = new WeakSet(); + /** Served payload object → the ETag of the document it came from. */ + private readonly servedPayloads = new WeakMap(); constructor(private readonly api: ServerApi) {} @@ -52,37 +54,46 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { } if (response.status === 304 && cached) { - return this.serve(cached.document); + return this.serve(cached.document, cached.etag); } const document = (await response.json()) as Playlist; - this.remember(playlistId, response, document); - return this.serve(document); + const etag = this.remember(playlistId, response, document); + return this.serve(document, etag); } async dbUpsertAppPlaylist( playlist: Playlist ): Promise { const path = `/playlists/${encodeURIComponent(playlist._id)}`; - const keepPayload = this.carriesServedPayload(playlist); - let response: Response; - try { - response = keepPayload - ? await this.api.sendJson( - 'PUT', - `${path}?keepPayload=1`, - withoutPayload(playlist) - ) - : await this.api.sendJson('PUT', path, playlist); - } catch (error) { - if (!( - keepPayload && - error instanceof ServerApiError && - error.code === 'payload_missing' - )) { - throw error; + const payloadEtag = this.servedPayloadEtag(playlist); + if (payloadEtag !== null) { + try { + const response = await this.api.sendJson( + 'PUT', + `${path}?keepPayload=1`, + withoutPayload(playlist) + ); + // The server kept its stored payload; it is ours only if + // nothing replaced it since we read it. + const replaced = response.headers.get( + SERVER_REPLACED_ETAG_HEADER + ); + if (replaced && sameEtag(replaced, payloadEtag)) { + this.remember(playlist._id, response, playlist); + } else { + this.cache.delete(playlist._id); + } + return { success: true }; + } catch (error) { + if (!( + error instanceof ServerApiError && + error.code === 'payload_missing' + )) { + throw error; + } } - response = await this.api.sendJson('PUT', path, playlist); } + const response = await this.api.sendJson('PUT', path, playlist); this.remember(playlist._id, response, playlist); return { success: true }; } @@ -131,43 +142,60 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { return { success: true }; } - private remember(id: string, response: Response, document: Playlist): void { + private remember( + id: string, + response: Response, + document: Playlist + ): string | null { const etag = response.headers.get('etag'); if (etag) { this.cache.set(id, { etag, document }); } else { this.cache.delete(id); } + return etag; } - private serve(document: Playlist): Playlist { + private serve(document: Playlist, etag: string | null): Playlist { const copy = structuredClone(document); + if (!etag) { + return copy; + } const record = copy as unknown as Record; for (const field of SERVER_PAYLOAD_FIELDS) { const value = record[field]; if (value && typeof value === 'object') { - this.servedPayloads.add(value); + this.servedPayloads.set(value, etag); } } return copy; } - private carriesServedPayload(playlist: Playlist): boolean { + /** The ETag every payload field of `playlist` was served under, if any. */ + private servedPayloadEtag(playlist: Playlist): string | null { const record = playlist as unknown as Record; - const present = SERVER_PAYLOAD_FIELDS.filter( - (field) => record[field] !== undefined && record[field] !== null - ); - return ( - present.length > 0 && - present.every( - (field) => - typeof record[field] === 'object' && - this.servedPayloads.has(record[field] as object) - ) - ); + const etags = new Set(); + for (const field of SERVER_PAYLOAD_FIELDS) { + const value = record[field]; + if (value === undefined || value === null) { + continue; + } + etags.add( + typeof value === 'object' + ? this.servedPayloads.get(value) + : undefined + ); + } + const [etag] = etags; + return etags.size === 1 && etag !== undefined ? etag : null; } } +/** nginx's gzip turns ETags weak; the revision stays the same. */ +function sameEtag(left: string, right: string): boolean { + return left.replace(/^W\//, '') === right.replace(/^W\//, ''); +} + function withoutPayload(playlist: Playlist): Record { const copy = { ...playlist } as Record; for (const field of SERVER_PAYLOAD_FIELDS) { diff --git a/libs/services/src/lib/server-storage/server-storage-session.ts b/libs/services/src/lib/server-storage/server-storage-session.ts index d964c28a2..ee1a85b50 100644 --- a/libs/services/src/lib/server-storage/server-storage-session.ts +++ b/libs/services/src/lib/server-storage/server-storage-session.ts @@ -38,7 +38,6 @@ export class ServerStorageSession { readonly userData = new ServerUserDataStorage( this.api, globalThis.localStorage, - () => this.backend.usesBrowserStorageOverride(), () => this.announceWriteFailure() ); 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 e0991df01..d72e33dc5 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 @@ -10,11 +10,18 @@ function memoryStorage() { }; } +/** Lets chained promises run without advancing the fake timers. */ +async function settle(): Promise { + for (let i = 0; i < 10; i += 1) { + await Promise.resolve(); + } +} + describe('ServerUserDataStorage', () => { beforeEach(() => jest.useFakeTimers()); afterEach(() => jest.useRealTimers()); - function setup(useLocal = false) { + function setup() { const local = memoryStorage(); const api = { getJson: jest.fn().mockResolvedValue({ @@ -33,7 +40,6 @@ describe('ServerUserDataStorage', () => { const storage = new ServerUserDataStorage( api as never, local, - () => useLocal, onWriteFailure ); return { api, local, storage, onWriteFailure }; @@ -70,13 +76,11 @@ describe('ServerUserDataStorage', () => { expect(api.sendJson).toHaveBeenCalledTimes(1); }); - it('passes other keys and the browser override to plain storage', () => { - const { api, local, storage } = setup(true); + it('passes other keys to plain storage', () => { + const { api, local, storage } = setup(); - storage.setItem('xtream-favorites', '{"legacy":[]}'); storage.setItem('xtream-playlists', '[]'); - expect(local.values.get('xtream-favorites')).toBe('{"legacy":[]}'); expect(local.values.get('xtream-playlists')).toBe('[]'); expect(api.sendJson).not.toHaveBeenCalled(); }); @@ -107,4 +111,84 @@ describe('ServerUserDataStorage', () => { expect(onWriteFailure).toHaveBeenCalled(); }); + + it('keeps an edit made while a refresh is reading', async () => { + const { api, storage } = setup(); + await storage.hydrate(); + let answer: (value: unknown) => void = () => undefined; + api.getJson.mockReturnValueOnce( + new Promise((resolve) => (answer = resolve)) + ); + + const refresh = storage.hydrate(); + await settle(); + storage.setItem( + 'xtream-favorites', + JSON.stringify({ p1: ['live:1'], p2: ['live:2'] }) + ); + // The server's older snapshot arrives after the edit. + answer({ entries: { 'xtream-favorites': { p1: ['live:1'] } } }); + await refresh; + + expect(JSON.parse(storage.getItem('xtream-favorites') ?? '')).toEqual({ + p1: ['live:1'], + p2: ['live:2'], + }); + await jest.runOnlyPendingTimersAsync(); + expect(api.sendJson).toHaveBeenCalledWith( + 'PUT', + '/user-state/xtream-favorites/p2', + { value: ['live:2'] } + ); + + // The next edit does not send a delete for the kept entry. + storage.setItem( + 'xtream-favorites', + JSON.stringify({ p1: ['live:1', 'live:3'], p2: ['live:2'] }) + ); + await jest.runOnlyPendingTimersAsync(); + expect(api.request).not.toHaveBeenCalledWith( + '/user-state/xtream-favorites/p2', + { method: 'DELETE' } + ); + }); + + it('sends a newer batch only after the older one finished', async () => { + const { api, storage } = setup(); + await storage.hydrate(); + const order: string[] = []; + let finishFirst: () => void = () => undefined; + api.sendJson + .mockImplementationOnce( + (_method: string, _path: string, body: { value: unknown }) => { + order.push(`start ${JSON.stringify(body.value)}`); + return new Promise((resolve) => { + finishFirst = () => { + order.push('end first'); + resolve(); + }; + }); + } + ) + .mockImplementation( + (_method: string, _path: string, body: { value: unknown }) => { + order.push(`start ${JSON.stringify(body.value)}`); + return Promise.resolve(); + } + ); + + storage.setItem('xtream-favorites', JSON.stringify({ p1: ['a'] })); + const first = storage.flush(); + await settle(); + // A newer edit while the first batch is still travelling. + storage.setItem('xtream-favorites', JSON.stringify({ p1: ['b'] })); + const second = storage.flush(); + await settle(); + expect(order).toEqual(['start ["a"]']); + + finishFirst(); + await first; + await second; + expect(order).toEqual(['start ["a"]', 'end first', 'start ["b"]']); + }); }); 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 e163b2657..d335f2b32 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 @@ -27,7 +27,7 @@ type ServerApi = Pick< * Write-through `USER_DATA_STORAGE` for server mode. Synced namespaces are * maps keyed by playlist id; a write compares the new map with the last one * the server saw and sends only the playlist entries that changed. Other - * keys, and every key while the browser override runs, use plain storage. + * keys use plain storage. */ export class ServerUserDataStorage implements UserDataStorage { private readonly synced = new Map< @@ -36,11 +36,15 @@ export class ServerUserDataStorage implements UserDataStorage { >(); private readonly pending = new Map(); private flushTimer: ReturnType | null = null; + /** Batches go out one at a time, so an older write never lands last. */ + private sending: Promise = Promise.resolve(); + /** Edit counter per `namespace|playlistId`, for hydrate to keep them. */ + private readonly edits = new Map(); + private editCount = 0; constructor( private readonly api: ServerApi, private readonly local: UserDataStorage, - private readonly useLocal: () => boolean, private readonly onWriteFailure: (error: unknown) => void ) {} @@ -68,24 +72,58 @@ export class ServerUserDataStorage implements UserDataStorage { this.queueChanges(key, {}); } - /** Replaces the cache with the server's data, after sending pending writes. */ + /** + * Replaces the cache with the server's data, after sending pending + * writes. Entries edited while the read was in flight keep their local + * value: their write is queued or travelling and will reach the server. + */ async hydrate(): Promise { + // Edits from here on may miss the read: queued, or still travelling. + const editsBefore = this.editCount; await this.flush(); const { entries } = await this.api.getJson('/user-state'); for (const namespace of SERVER_USER_STATE_NAMESPACES) { - const value = entries[namespace] ?? {}; + const value: Record = { + ...(entries[namespace] ?? {}), + }; + const current = parseMap( + this.local.getItem(SERVER_CACHE_PREFIX + namespace) ?? '{}' + ); + for (const [key, edit] of this.edits) { + const [editedNamespace, playlistId] = splitEditKey(key); + if (edit <= editsBefore || editedNamespace !== namespace) { + continue; + } + if (playlistId in current) { + value[playlistId] = current[playlistId]; + } else { + delete value[playlistId]; + } + } this.synced.set(namespace, value); this.local.setItem( SERVER_CACHE_PREFIX + namespace, JSON.stringify(value) ); } + for (const [key, edit] of this.edits) { + if (edit <= editsBefore) { + this.edits.delete(key); + } + } } - async flush(): Promise { - const batch = this.takePending(); - await Promise.all(batch.map((entry) => this.send(entry))); + /** Sends the queued entries after any batch still travelling. */ + flush(): Promise { + const run = this.sending.then(() => + Promise.all(this.takePending().map((entry) => this.send(entry))) + ); + this.sending = run.then( + () => undefined, + () => undefined + ); + return this.sending; } flushWithKeepalive(): void { @@ -100,7 +138,7 @@ export class ServerUserDataStorage implements UserDataStorage { } private isSynced(key: string): key is ServerUserStateNamespace { - return !this.useLocal() && isServerUserStateNamespace(key); + return isServerUserStateNamespace(key); } private queueChanges( @@ -112,26 +150,24 @@ export class ServerUserDataStorage implements UserDataStorage { if ( JSON.stringify(previous[playlistId]) !== JSON.stringify(value) ) { - this.pending.set(`${namespace}|${playlistId}`, { - namespace, - playlistId, - value, - }); + this.queue({ namespace, playlistId, value }); } } for (const playlistId of Object.keys(previous)) { if (!(playlistId in next)) { - this.pending.set(`${namespace}|${playlistId}`, { - namespace, - playlistId, - value: undefined, - }); + this.queue({ namespace, playlistId, value: undefined }); } } this.synced.set(namespace, next); this.scheduleFlush(); } + private queue(entry: PendingEntry): void { + const key = `${entry.namespace}|${entry.playlistId}`; + this.pending.set(key, entry); + this.edits.set(key, ++this.editCount); + } + private scheduleFlush(): void { if (this.flushTimer !== null || this.pending.size === 0) { return; @@ -167,6 +203,11 @@ export class ServerUserDataStorage implements UserDataStorage { } } +function splitEditKey(key: string): [string, string] { + const separator = key.indexOf('|'); + return [key.slice(0, separator), key.slice(separator + 1)]; +} + function entryPath(entry: PendingEntry): string { return `/user-state/${entry.namespace}/${encodeURIComponent(entry.playlistId)}`; } diff --git a/libs/shared/interfaces/src/lib/server-storage.contract.ts b/libs/shared/interfaces/src/lib/server-storage.contract.ts index e4af9e3b7..a10b46376 100644 --- a/libs/shared/interfaces/src/lib/server-storage.contract.ts +++ b/libs/shared/interfaces/src/lib/server-storage.contract.ts @@ -36,6 +36,13 @@ export interface ServerInstanceInfo { readonly auth: 'none' | 'token'; } +/** + * Sent with every `PUT /v1/playlists/:id`: the document's ETag before that + * write. A client that kept the stored payload (`keepPayload=1`) may cache + * its copy under the new ETag only if its payload came from this one. + */ +export const SERVER_REPLACED_ETAG_HEADER = 'X-Replaced-ETag'; + export interface ServerStorageStatusResponse { readonly playlists: number; readonly databaseBytes: number;