From 6820b87005ce0bcf83733be72da198b588bdf566 Mon Sep 17 00:00:00 2001 From: 4gray Date: Mon, 12 Oct 2026 00:29:55 +0200 Subject: [PATCH] fix(server-storage): positions by content, acknowledged keepalive, serial re-reads, token changes - Moved playback positions are matched by content type and Xtream id and name the server playlist. - Page-hide writes report whether the server took them; refused ones go back into the queue. Re-reads run one at a time. - The access guard's 401 carries WWW-Authenticate: Bearer realm="iptvnator"; the interceptor and PwaService's raw fetches turn it into sign-in, so a token changed on the server is noticed on provider calls too. Co-Authored-By: Claude Opus 5.5 --- .../src/app/storage/access-guard.spec.ts | 7 ++ .../src/app/storage/access-guard.ts | 14 ++-- apps/web/src/app/services/pwa.service.spec.ts | 42 ++++++++++ apps/web/src/app/services/pwa.service.ts | 23 +++++- docs/architecture/pwa-self-hosted.md | 20 +++-- libs/services/src/lib/server-access-token.ts | 10 +++ .../src/lib/server-storage-status.service.ts | 12 +++ .../server-access-token.interceptor.spec.ts | 38 ++++++++++ .../server-access-token.interceptor.ts | 41 ++++++++-- .../server-storage/server-api-client.spec.ts | 15 ++++ .../lib/server-storage/server-api-client.ts | 35 +++++---- .../server-migration-merge.spec.ts | 49 +++++++++++- .../server-storage/server-migration-merge.ts | 31 +++++++- .../server-migration.service.ts | 8 +- .../server-user-data-storage.spec.ts | 42 +++++++++- .../server-user-data-storage.ts | 76 ++++++++++++++----- .../src/lib/server-storage.contract.ts | 6 ++ 17 files changed, 414 insertions(+), 55 deletions(-) diff --git a/apps/web-backend/src/app/storage/access-guard.spec.ts b/apps/web-backend/src/app/storage/access-guard.spec.ts index 49e90b308..92d2b69aa 100644 --- a/apps/web-backend/src/app/storage/access-guard.spec.ts +++ b/apps/web-backend/src/app/storage/access-guard.spec.ts @@ -30,6 +30,13 @@ describe('createAccessGuard', () => { code: 'unauthorized', }); expect(wrong.status).toBe(401); + // Lets the PWA tell it from a provider's 401 on a proxy route. + expect(wrong.headers.get('www-authenticate')).toBe( + 'Bearer realm="iptvnator"' + ); + expect(wrong.headers.get('access-control-expose-headers')).toBe( + 'WWW-Authenticate' + ); })); it('accepts the right token and lets preflight through', () => diff --git a/apps/web-backend/src/app/storage/access-guard.ts b/apps/web-backend/src/app/storage/access-guard.ts index 084cc3d10..a8068069e 100644 --- a/apps/web-backend/src/app/storage/access-guard.ts +++ b/apps/web-backend/src/app/storage/access-guard.ts @@ -1,5 +1,6 @@ import { createHash, timingSafeEqual } from 'node:crypto'; import type { RequestHandler } from 'express'; +import { SERVER_ACCESS_CHALLENGE } from '@iptvnator/shared/interfaces'; /** Routes a browser must reach before it has a token. */ const PUBLIC_PATHS = new Set(['/', '/health', '/config.js', '/v1/instance']); @@ -24,11 +25,14 @@ export function createAccessGuard(token: string): RequestHandler { return; } - res.status(401).json({ - message: 'This IPTVnator server requires an access token.', - status: 401, - code: 'unauthorized', - }); + res.status(401) + .set('WWW-Authenticate', SERVER_ACCESS_CHALLENGE) + .set('Access-Control-Expose-Headers', 'WWW-Authenticate') + .json({ + message: 'This IPTVnator server requires an access token.', + status: 401, + code: 'unauthorized', + }); }; } diff --git a/apps/web/src/app/services/pwa.service.spec.ts b/apps/web/src/app/services/pwa.service.spec.ts index e3bc14800..0238e94ae 100644 --- a/apps/web/src/app/services/pwa.service.spec.ts +++ b/apps/web/src/app/services/pwa.service.spec.ts @@ -20,6 +20,7 @@ import { getStalkerRequestErrorStatus, isStalkerProbeTimeout, } from '@iptvnator/portal/stalker/data-access'; +import { ServerStorageStatusService } from '@iptvnator/services'; import { PwaService } from './pwa.service'; describe('PwaService', () => { @@ -279,6 +280,47 @@ describe('PwaService', () => { globalThis.fetch = originalFetch; }); + it('asks for a new server token when the access guard refuses a Stalker call', async () => { + const status = TestBed.inject(ServerStorageStatusService); + status.activate( + { + service: 'iptvnator', + apiVersion: 1, + instanceId: 'i', + storage: 'server', + auth: 'token', + }, + () => undefined + ); + const originalFetch = globalThis.fetch; + globalThis.fetch = jest.fn().mockResolvedValue({ + ok: false, + status: 401, + statusText: 'Unauthorized', + headers: new Headers({ + 'WWW-Authenticate': 'Bearer realm="iptvnator"', + }), + json: async () => ({ code: 'unauthorized' }), + } as unknown as Response) as unknown as typeof fetch; + + const outcome = ( + service.sendIpcEvent(STALKER_REQUEST, { + url: 'http://portal.example/portal.php', + macAddress: '00:1A:79:AA:BB:CC', + params: { action: 'get_genres' }, + silent: true, + }) as Promise + ).catch((error: unknown) => error); + await Promise.resolve(); + http.expectOne((candidate) => + candidate.url.endsWith('/provider-targets') + ).flush({ targetId: 'target-1' }); + await outcome; + + expect(status.state()).toBe('unauthorized'); + globalThis.fetch = originalFetch; + }); + it('sends Stalker credentials as /stalker control params, including the serial', async () => { // JSDOM ships no fetch; install one for the proxy call. const fetchMock = jest.fn().mockResolvedValue({ diff --git a/apps/web/src/app/services/pwa.service.ts b/apps/web/src/app/services/pwa.service.ts index e88da7050..52c63b94a 100644 --- a/apps/web/src/app/services/pwa.service.ts +++ b/apps/web/src/app/services/pwa.service.ts @@ -13,7 +13,12 @@ import { switchMap, throwError, } from 'rxjs'; -import { DataService, serverAuthorizationHeaders } from '@iptvnator/services'; +import { + DataService, + isServerAccessChallenge, + serverAuthorizationHeaders, + ServerStorageStatusService, +} from '@iptvnator/services'; import { CONNECTIVITY_GUARD_RESET, ERROR, @@ -83,6 +88,7 @@ export class PwaService extends DataService { private readonly store = inject(Store); private readonly swUpdate = inject(SwUpdate); private readonly translateService = inject(TranslateService); + private readonly serverStatus = inject(ServerStorageStatusService); private readonly logger = createLogger('PwaService'); private readonly providerTargetIds = new Map>(); private readonly silentXtreamActions = new Set([ @@ -207,6 +213,7 @@ export class PwaService extends DataService { signal: controller.signal, } ); + this.noteServerAccess(response); if (!response.ok) { return { reset: false }; @@ -643,6 +650,7 @@ export class PwaService extends DataService { const response = await fetch(requestUrl, { headers: serverAuthorizationHeaders(), }); + this.noteServerAccess(response); if (!response.ok) { throw new Error( @@ -794,4 +802,17 @@ export class PwaService extends DataService { getAppEnvironment(): string { return 'pwa'; } + + /** A token changed on the server shows up on any backend call. */ + private noteServerAccess(response: Response): void { + if ( + response.status === 401 && + isServerAccessChallenge( + response.status, + response.headers.get('WWW-Authenticate') + ) + ) { + this.serverStatus.reportAccessDenied(); + } + } } diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index d81ad9865..8f832cfb7 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -104,7 +104,9 @@ its database. Storage errors are logged by error name and code only, because Drizzle messages can carry bound values such as portal credentials. With `IPTVNATOR_ACCESS_TOKEN`, every route except `/`, `/health`, `/config.js` and `/v1/instance` requires `Authorization: Bearer `, the proxy -routes included; the PWA sends it once its sign-in page has stored it. +routes included; its 401 carries `WWW-Authenticate: Bearer +realm="iptvnator"`, which a provider error passed on by the proxy never +does; the PWA sends it once its sign-in page has stored it. Browser app state written through `/v1/app-state` is stored under a `client:` key prefix, apart from server-owned keys such as the instance id. @@ -197,9 +199,10 @@ settings and guards read them in every build. 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. Batches go out one after another - (the page-hide flush waits for a batch still travelling), a failed write - stays queued for the next flush, and a re-read keeps entries edited while - it was in flight or not yet saved. Start-up waits + (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 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 @@ -215,7 +218,9 @@ settings and guards read them in every build. 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. + backend, and `PwaService` adds it to its two raw `fetch` calls. Both turn + the guard's 401 (`isServerAccessChallenge`) into `reportAccessDenied()`, + so a token changed on the server sends an open page back to sign-in. - **Moving browser data.** `ServerMigrationService` offers, once per server instance (marker `iptvnator-server:migration:`), to move the playlists this browser kept in IndexedDB (`planServerMigration`). A @@ -223,8 +228,9 @@ settings and guards read them in every build. 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 - the server id; `combineUserData` unions lists and merges maps, the - server's value winning on a conflict. Browser data is never deleted, and + 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 the marker is written only when nothing failed, so Settings → Server keeps offering a partial move. diff --git a/libs/services/src/lib/server-access-token.ts b/libs/services/src/lib/server-access-token.ts index 21f536443..53ed9c27f 100644 --- a/libs/services/src/lib/server-access-token.ts +++ b/libs/services/src/lib/server-access-token.ts @@ -1,5 +1,15 @@ +import { SERVER_ACCESS_CHALLENGE } from '@iptvnator/shared/interfaces'; + export const SERVER_ACCESS_TOKEN_STORAGE_KEY = 'iptvnator-server:access-token'; +/** A 401 from the backend's access guard, not one a provider sent. */ +export function isServerAccessChallenge( + status: number, + challenge: string | null +): boolean { + return status === 401 && challenge === SERVER_ACCESS_CHALLENGE; +} + export function readServerAccessToken(): string | null { try { const token = localStorage diff --git a/libs/services/src/lib/server-storage-status.service.ts b/libs/services/src/lib/server-storage-status.service.ts index 4ae53f73c..819c1fd72 100644 --- a/libs/services/src/lib/server-storage-status.service.ts +++ b/libs/services/src/lib/server-storage-status.service.ts @@ -1,5 +1,6 @@ import { computed, Injectable, signal } from '@angular/core'; import type { ServerInstanceInfo } from '@iptvnator/shared/interfaces'; +import { clearServerAccessToken } from './server-access-token'; export type ServerConnectionState = 'inactive' | 'online' | 'offline' | 'unauthorized'; @@ -49,6 +50,17 @@ export class ServerStorageStatusService { } } + /** + * The access guard refused a request outside the storage API (a + * provider proxy call): the stored token is stale, ask for a new one. + */ + reportAccessDenied(): void { + if (this.isActive()) { + clearServerAccessToken(); + this.stateSignal.set('unauthorized'); + } + } + retry(): void { this.retryHandler?.(); } diff --git a/libs/services/src/lib/server-storage/server-access-token.interceptor.spec.ts b/libs/services/src/lib/server-storage/server-access-token.interceptor.spec.ts index 689f66242..ea6fd8251 100644 --- a/libs/services/src/lib/server-storage/server-access-token.interceptor.spec.ts +++ b/libs/services/src/lib/server-storage/server-access-token.interceptor.spec.ts @@ -10,6 +10,7 @@ import { } from '@angular/common/http/testing'; import { TestBed } from '@angular/core/testing'; import { SERVER_ACCESS_TOKEN_STORAGE_KEY } from '../server-access-token'; +import { ServerStorageStatusService } from '../server-storage-status.service'; import { ServerAccessTokenInterceptor } from './server-access-token.interceptor'; import { SERVER_STORAGE_CONFIG } from './server-storage.config'; @@ -53,4 +54,41 @@ describe('ServerAccessTokenInterceptor', () => { .request.headers.has('Authorization') ).toBe(false); }); + + it('asks for a new token when the backend refuses the stored one', () => { + const status = TestBed.inject(ServerStorageStatusService); + status.activate( + { + service: 'iptvnator', + apiVersion: 1, + instanceId: 'i', + storage: 'server', + auth: 'token', + }, + () => undefined + ); + const http = TestBed.inject(HttpClient); + const controller = TestBed.inject(HttpTestingController); + + // A provider's own 401 passed on by the proxy changes nothing. + http.get('/api/xtream').subscribe({ error: () => undefined }); + controller + .expectOne('/api/xtream') + .flush({ message: 'bad login' }, { status: 401, statusText: 'x' }); + expect(status.state()).toBe('online'); + + http.get('/api/xtream').subscribe({ error: () => undefined }); + controller.expectOne('/api/xtream').flush( + { code: 'unauthorized' }, + { + status: 401, + statusText: 'Unauthorized', + headers: { 'WWW-Authenticate': 'Bearer realm="iptvnator"' }, + } + ); + expect(status.state()).toBe('unauthorized'); + expect( + localStorage.getItem(SERVER_ACCESS_TOKEN_STORAGE_KEY) + ).toBeNull(); + }); }); diff --git a/libs/services/src/lib/server-storage/server-access-token.interceptor.ts b/libs/services/src/lib/server-storage/server-access-token.interceptor.ts index a8d390a84..609814ba7 100644 --- a/libs/services/src/lib/server-storage/server-access-token.interceptor.ts +++ b/libs/services/src/lib/server-storage/server-access-token.interceptor.ts @@ -1,17 +1,27 @@ import { + HttpErrorResponse, HttpEvent, HttpHandler, HttpInterceptor, HttpRequest, } from '@angular/common/http'; import { inject, Injectable } from '@angular/core'; -import { Observable } from 'rxjs'; -import { readServerAccessToken } from '../server-access-token'; +import { Observable, tap } from 'rxjs'; +import { + isServerAccessChallenge, + readServerAccessToken, +} from '../server-access-token'; +import { ServerStorageStatusService } from '../server-storage-status.service'; import { SERVER_STORAGE_CONFIG } from './server-storage.config'; -/** Adds the server access token to `HttpClient` calls to the backend. */ +/** + * Adds the server access token to `HttpClient` calls to the backend, and + * sends the user to sign-in when the backend refuses it (a token changed + * on the server while this page was open). + */ @Injectable() export class ServerAccessTokenInterceptor implements HttpInterceptor { + private readonly status = inject(ServerStorageStatusService); private readonly baseUrl = inject(SERVER_STORAGE_CONFIG).baseUrl.replace( /\/+$/, '' @@ -21,12 +31,29 @@ export class ServerAccessTokenInterceptor implements HttpInterceptor { request: HttpRequest, next: HttpHandler ): Observable> { - const token = readServerAccessToken(); - if (!token || !request.url.startsWith(this.baseUrl)) { + if (!request.url.startsWith(this.baseUrl)) { return next.handle(request); } - return next.handle( - request.clone({ setHeaders: { Authorization: `Bearer ${token}` } }) + const token = readServerAccessToken(); + const sent = token + ? request.clone({ + setHeaders: { Authorization: `Bearer ${token}` }, + }) + : request; + return next.handle(sent).pipe( + tap({ + error: (error: unknown) => { + if ( + error instanceof HttpErrorResponse && + isServerAccessChallenge( + error.status, + error.headers.get('WWW-Authenticate') + ) + ) { + this.status.reportAccessDenied(); + } + }, + }) ); } } 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 0f9843631..0869289b3 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 @@ -84,6 +84,21 @@ describe('ServerApiClient', () => { expect(status.reportOffline).not.toHaveBeenCalled(); }); + it('reports whether a keepalive write was taken', async () => { + const ok = jest.fn().mockResolvedValue(response(200, {})); + await expect( + client(ok).sendKeepalive('PUT', '/user-state/a/b', { value: 1 }) + ).resolves.toBe(true); + expect(ok.mock.calls[0][1]).toMatchObject({ keepalive: true }); + + const refused = jest + .fn() + .mockRejectedValue(new TypeError('body too large')); + await expect( + client(refused).sendKeepalive('DELETE', '/user-state/a/b') + ).resolves.toBe(false); + }); + 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 71b89645f..774b41807 100644 --- a/libs/services/src/lib/server-storage/server-api-client.ts +++ b/libs/services/src/lib/server-storage/server-api-client.ts @@ -123,22 +123,31 @@ export class ServerApiClient { } /** Best effort while the page is hidden or closed; errors are ignored. */ - sendKeepalive( + /** + * A write that survives the page. Resolves whether the server took it; + * browsers reject keepalive bodies above 64 KiB, which resolves false. + */ + async sendKeepalive( method: 'PUT' | 'DELETE', path: string, body?: unknown - ): void { - void this.fetchFn()(this.url(path), { - method, - keepalive: true, - headers: { - ...serverAuthorizationHeaders(), - ...(body === undefined - ? {} - : { 'Content-Type': 'application/json' }), - }, - ...(body === undefined ? {} : { body: JSON.stringify(body) }), - }).catch(() => undefined); + ): Promise { + try { + const response = await this.fetchFn()(this.url(path), { + method, + keepalive: true, + headers: { + ...serverAuthorizationHeaders(), + ...(body === undefined + ? {} + : { 'Content-Type': 'application/json' }), + }, + ...(body === undefined ? {} : { body: JSON.stringify(body) }), + }); + return response.ok; + } catch { + return false; + } } private fetchFn(): typeof fetch { 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 5b0a1c1cc..fb1b8a03e 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,4 +1,8 @@ -import { combineUserData, unionItems } from './server-migration-merge'; +import { + combineUserData, + retargetUserData, + unionItems, +} from './server-migration-merge'; describe('combineUserData', () => { it('unions lists, server entries first', () => { @@ -40,3 +44,46 @@ describe('unionItems', () => { ]); }); }); + +describe('playback positions', () => { + it('keep the server position of the same item', () => { + expect( + combineUserData([ + [ + { + contentType: 'vod', + contentXtreamId: 7, + positionSeconds: 60, + }, + ], + [ + { + contentType: 'vod', + contentXtreamId: 7, + positionSeconds: 900, + }, + { + contentType: 'episode', + contentXtreamId: 7, + positionSeconds: 5, + }, + ], + ]) + ).toEqual([ + { contentType: 'vod', contentXtreamId: 7, positionSeconds: 60 }, + { contentType: 'episode', contentXtreamId: 7, positionSeconds: 5 }, + ]); + }); + + it('name the server playlist after a move', () => { + expect( + retargetUserData( + [{ contentXtreamId: 1, playlistId: 'b' }, 'live:1'], + 'b', + 's' + ) + ).toEqual([{ contentXtreamId: 1, playlistId: 's' }, 'live:1']); + const same = [{ playlistId: 'b' }]; + expect(retargetUserData(same, 'b', 'b')).toBe(same); + }); +}); 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 e1771110c..1fbfe5e87 100644 --- a/libs/services/src/lib/server-storage/server-migration-merge.ts +++ b/libs/services/src/lib/server-storage/server-migration-merge.ts @@ -41,11 +41,21 @@ export function unionItems(...lists: readonly (readonly T[])[]): T[] { return result; } -/** Content ids or URLs identify favorites and history entries. */ +/** + * Content ids or URLs identify favorites and history entries; a playback + * position is identified by its content type and Xtream id. + */ function itemKey(item: unknown): string { if (!isPlainObject(item)) { return `value:${JSON.stringify(item)}`; } + if ( + typeof item['contentType'] === 'string' && + (typeof item['contentXtreamId'] === 'number' || + typeof item['contentXtreamId'] === 'string') + ) { + return `position:${item['contentType']}:${item['contentXtreamId']}`; + } for (const field of ['id', 'stream_id', 'series_id', 'movie_id', 'url']) { const value = item[field]; if (typeof value === 'string' || typeof value === 'number') { @@ -55,6 +65,25 @@ function itemKey(item: unknown): string { return `json:${JSON.stringify(item)}`; } +/** + * A browser value moved under another playlist id: list items that name + * their playlist (playback positions do) name the server's instead. + */ +export function retargetUserData( + value: unknown, + browserId: string, + serverId: string +): unknown { + if (browserId === serverId || !Array.isArray(value)) { + return value; + } + return value.map((item) => + isPlainObject(item) && item['playlistId'] === browserId + ? { ...item, playlistId: serverId } + : item + ); +} + function isPlainObject(value: unknown): value is Record { return !!value && typeof value === 'object' && !Array.isArray(value); } 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 99b2d21a2..1654d0c2e 100644 --- a/libs/services/src/lib/server-storage/server-migration.service.ts +++ b/libs/services/src/lib/server-storage/server-migration.service.ts @@ -7,7 +7,11 @@ import { type Playlist, type ServerUserStateResponse, } from '@iptvnator/shared/interfaces'; -import { combineUserData, unionItems } from './server-migration-merge'; +import { + combineUserData, + retargetUserData, + unionItems, +} from './server-migration-merge'; import { planServerMigration } from './server-migration-plan'; import { SERVER_STORAGE_CONFIG } from './server-storage.config'; import { ServerStorageSession } from './server-storage-session'; @@ -180,7 +184,7 @@ export class ServerMigrationService { if (serverId) { incoming.set(serverId, [ ...(incoming.get(serverId) ?? []), - value, + retargetUserData(value, browserId, serverId), ]); } } 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 57546adb4..b75fc7bc5 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 @@ -34,7 +34,7 @@ describe('ServerUserDataStorage', () => { }), sendJson: jest.fn().mockResolvedValue(undefined), request: jest.fn().mockResolvedValue(undefined), - sendKeepalive: jest.fn(), + sendKeepalive: jest.fn().mockResolvedValue(true), }; const onWriteFailure = jest.fn(); const storage = new ServerUserDataStorage( @@ -259,4 +259,44 @@ describe('ServerUserDataStorage', () => { { value: ['b'] } ); }); + + it('queues a page-hide write again when the server did not take it', async () => { + const { api, storage } = setup(); + await storage.hydrate(); + api.sendKeepalive.mockResolvedValueOnce(false); + + storage.setItem('xtream-favorites', JSON.stringify({ p1: ['big'] })); + storage.flushWithKeepalive(); + await settle(); + + // Back on the page: the re-read sends it normally and keeps it. + api.getJson.mockResolvedValueOnce({ + entries: { 'xtream-favorites': { p1: ['big'] } }, + }); + await storage.hydrate(); + expect(api.sendJson).toHaveBeenCalledWith( + 'PUT', + '/user-state/xtream-favorites/p1', + { value: ['big'] } + ); + }); + + it('runs re-reads one after another', async () => { + const { api, storage } = setup(); + await storage.hydrate(); + let answerFirst: (value: unknown) => void = () => undefined; + api.getJson.mockReturnValueOnce( + new Promise((resolve) => (answerFirst = resolve)) + ); + + const first = storage.hydrate(); + const second = storage.hydrate(); + await settle(); + expect(api.getJson).toHaveBeenCalledTimes(2); + + answerFirst({ entries: {} }); + await first; + await second; + expect(api.getJson).toHaveBeenCalledTimes(3); + }); }); 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 136e4a73f..1006eccc1 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 @@ -39,6 +39,10 @@ export class ServerUserDataStorage implements UserDataStorage { /** Batches go out one at a time, so an older write never lands last. */ private sending: Promise = Promise.resolve(); private batchesInFlight = 0; + /** Page-hide writes still travelling, by `namespace|playlistId`. */ + private readonly keepaliveKeys = new Set(); + /** Re-reads run one at a time, so an older one never lands last. */ + private hydration: Promise = Promise.resolve(); /** Edit counter per `namespace|playlistId`, for hydrate to keep them. */ private readonly edits = new Map(); private editCount = 0; @@ -78,7 +82,16 @@ export class ServerUserDataStorage implements UserDataStorage { * writes. Entries edited while the read was in flight, and entries whose * write failed, keep their local value: the server does not have it yet. */ - async hydrate(): Promise { + hydrate(): Promise { + const run = this.hydration.then(() => this.hydrateNow()); + this.hydration = run.then( + () => undefined, + () => undefined + ); + return run; + } + + private async hydrateNow(): Promise { // Edits from here on may miss the read: queued, or still travelling. const editsBefore = this.editCount; await this.flush(); @@ -91,7 +104,10 @@ export class ServerUserDataStorage implements UserDataStorage { const current = parseMap( this.local.getItem(SERVER_CACHE_PREFIX + namespace) ?? '{}' ); - const unsaved = new Set(this.pending.keys()); + const unsaved = new Set([ + ...this.pending.keys(), + ...this.keepaliveKeys, + ]); for (const [key, edit] of this.edits) { if (edit > editsBefore) { unsaved.add(key); @@ -115,7 +131,11 @@ export class ServerUserDataStorage implements UserDataStorage { ); } for (const [key, edit] of this.edits) { - if (edit <= editsBefore) { + if ( + edit <= editsBefore && + !this.pending.has(key) && + !this.keepaliveKeys.has(key) + ) { this.edits.delete(key); } } @@ -153,14 +173,22 @@ export class ServerUserDataStorage implements UserDataStorage { } const sendAll = () => { for (const entry of entries) { + const key = entryKey(entry); + const edit = this.edits.get(key); + this.keepaliveKeys.add(key); const path = entryPath(entry); - if (entry.value === undefined) { - this.api.sendKeepalive('DELETE', path); - } else { - this.api.sendKeepalive('PUT', path, { - value: entry.value, - }); - } + const sent = + entry.value === undefined + ? this.api.sendKeepalive('DELETE', path) + : this.api.sendKeepalive('PUT', path, { + value: entry.value, + }); + void sent.then((ok) => { + this.keepaliveKeys.delete(key); + if (!ok) { + this.requeue(entry, edit); + } + }); } }; if (this.batchesInFlight === 0) { @@ -196,7 +224,7 @@ export class ServerUserDataStorage implements UserDataStorage { } private queue(entry: PendingEntry): void { - const key = `${entry.namespace}|${entry.playlistId}`; + const key = entryKey(entry); this.pending.set(key, entry); this.edits.set(key, ++this.editCount); } @@ -221,7 +249,22 @@ export class ServerUserDataStorage implements UserDataStorage { return batch; } + /** + * 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 { + const key = entryKey(entry); + if (this.edits.get(key) === editWhenSent && !this.pending.has(key)) { + this.pending.set(key, 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' }); @@ -231,17 +274,16 @@ export class ServerUserDataStorage implements UserDataStorage { }); } } catch (error) { - // Keep it for the next flush (the offline retry or the next - // edit) unless a newer value for the entry is already queued. - const key = `${entry.namespace}|${entry.playlistId}`; - if (!this.pending.has(key)) { - this.pending.set(key, entry); - } + this.requeue(entry, edit); this.onWriteFailure(error); } } } +function entryKey(entry: PendingEntry): string { + return `${entry.namespace}|${entry.playlistId}`; +} + function splitEditKey(key: string): [string, string] { const separator = key.indexOf('|'); return [key.slice(0, separator), key.slice(separator + 1)]; diff --git a/libs/shared/interfaces/src/lib/server-storage.contract.ts b/libs/shared/interfaces/src/lib/server-storage.contract.ts index a10b46376..a9364be61 100644 --- a/libs/shared/interfaces/src/lib/server-storage.contract.ts +++ b/libs/shared/interfaces/src/lib/server-storage.contract.ts @@ -36,6 +36,12 @@ export interface ServerInstanceInfo { readonly auth: 'none' | 'token'; } +/** + * `WWW-Authenticate` value of the access guard's 401. A provider error the + * proxy passes on never carries it, so a client can tell the two apart. + */ +export const SERVER_ACCESS_CHALLENGE = 'Bearer realm="iptvnator"'; + /** * 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