diff --git a/apps/web-backend/src/app/storage/server-user-state-repository.ts b/apps/web-backend/src/app/storage/server-user-state-repository.ts index a61b0de2f..732526ffe 100644 --- a/apps/web-backend/src/app/storage/server-user-state-repository.ts +++ b/apps/web-backend/src/app/storage/server-user-state-repository.ts @@ -1,16 +1,88 @@ +import { createHash } from 'node:crypto'; import { and, eq } from 'drizzle-orm'; import type { DatabaseInstance } from '@iptvnator/shared/database'; import { SERVER_USER_STATE_NAMESPACES, + type ServerUserStateCurrent, type ServerUserStateNamespace, type ServerUserStateResponse, } from '@iptvnator/shared/interfaces'; import { serverUserState } from './server-user-state.table'; +/** + * What a conditional write is based on: `undefined` writes unconditionally, + * `null` expects no entry, a string expects that tag. + */ +export type ExpectedTag = string | null | undefined; + +export type ConditionalWriteResult = + | { readonly tag: string | null } + | { readonly conflict: ServerUserStateCurrent }; + +/** Version tag of a stored value: its content, so equal values match. */ +export function userStateTag(serialized: string): string { + return createHash('sha256').update(serialized).digest('hex').slice(0, 16); +} + /** Xtream PWA user data, one JSON value per namespace and playlist. */ export class ServerUserStateRepository { constructor(private readonly db: DatabaseInstance) {} + /** Every entry with its tag, as `GET /v1/user-state` serves them. */ + allWithTags(): Required { + const tags = Object.fromEntries( + SERVER_USER_STATE_NAMESPACES.map((namespace) => [ + namespace, + Object.create(null) as Record, + ]) + ) as Required['tags']; + for (const row of this.db.select().from(serverUserState).all()) { + const bucket = tags[row.namespace as ServerUserStateNamespace]; + if (bucket) { + bucket[row.playlistId] = userStateTag(row.value); + } + } + return { entries: this.all(), tags }; + } + + /** + * Writes (`value`) or deletes (`undefined`) one entry if it still has + * the expected tag; otherwise reports the current entry. Synchronous on + * the single connection, so the check and the write cannot interleave. + */ + writeIf( + namespace: ServerUserStateNamespace, + playlistId: string, + value: unknown, + expected: ExpectedTag + ): ConditionalWriteResult { + if (expected !== undefined) { + const row = this.db + .select({ value: serverUserState.value }) + .from(serverUserState) + .where( + and( + eq(serverUserState.namespace, namespace), + eq(serverUserState.playlistId, playlistId) + ) + ) + .get(); + const tag = row ? userStateTag(row.value) : null; + if (tag !== expected) { + return { + conflict: row + ? { value: JSON.parse(row.value), tag } + : { tag: null }, + }; + } + } + if (value === undefined) { + this.delete(namespace, playlistId); + return { tag: null }; + } + return { tag: this.put(namespace, playlistId, value) }; + } + all(): ServerUserStateResponse['entries'] { // Prototype-free buckets: a playlist id such as `__proto__` must be // stored as an own key, not change the bucket's prototype. @@ -30,11 +102,12 @@ export class ServerUserStateRepository { return entries; } + /** Returns the stored value's tag. */ put( namespace: ServerUserStateNamespace, playlistId: string, value: unknown - ): void { + ): string { const serialized = JSON.stringify(value ?? null); const updatedAt = new Date().toISOString(); this.db @@ -45,6 +118,7 @@ export class ServerUserStateRepository { set: { value: serialized, updatedAt }, }) .run(); + return userStateTag(serialized); } delete(namespace: ServerUserStateNamespace, playlistId: string): void { 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 d96c1c65f..83b354491 100644 --- a/apps/web-backend/src/app/storage/storage-routes.spec.ts +++ b/apps/web-backend/src/app/storage/storage-routes.spec.ts @@ -205,6 +205,75 @@ describe('server storage routes', () => { }) )); + it('writes user state only on the tag it was read at', () => + withTempStorage((storage) => + withServer(app(storage), async (baseUrl) => { + const entry = `${baseUrl}/v1/user-state/xtream-favorites/p1`; + // Created only while absent. + const created = await fetch( + entry, + json( + { value: ['live:1'] }, + { method: 'PUT', headers: { 'if-none-match': '*' } } + ) + ); + const { tag } = (await created.json()) as { tag: string }; + expect(tag).toMatch(/^[0-9a-f]{16}$/); + const again = await fetch( + entry, + json( + { value: ['live:2'] }, + { method: 'PUT', headers: { 'if-none-match': '*' } } + ) + ); + expect(again.status).toBe(412); + await expect(again.json()).resolves.toMatchObject({ + code: 'conflict', + current: { value: ['live:1'], tag }, + }); + + const read = (await ( + await fetch(`${baseUrl}/v1/user-state`) + ).json()) as ServerUserStateResponse; + expect(read.tags?.['xtream-favorites']['p1']).toBe(tag); + + // Another device wrote meanwhile: a stale tag is refused. + const next = await fetch( + entry, + json( + { value: ['live:1', 'live:9'] }, + { method: 'PUT', headers: { 'if-match': `"${tag}"` } } + ) + ); + const nextTag = ((await next.json()) as { tag: string }).tag; + const stale = await fetch( + entry, + json( + { value: ['live:1', 'live:5'] }, + { method: 'PUT', headers: { 'if-match': `"${tag}"` } } + ) + ); + expect(stale.status).toBe(412); + await expect(stale.json()).resolves.toMatchObject({ + current: { value: ['live:1', 'live:9'], tag: nextTag }, + }); + + const removed = await fetch(entry, { + method: 'DELETE', + headers: { 'if-match': `"${nextTag}"` }, + }); + await expect(removed.json()).resolves.toMatchObject({ + tag: null, + }); + // Unconditional writes keep working for older clients. + const plain = await fetch( + entry, + json({ value: ['x'] }, { method: 'PUT' }) + ); + expect(plain.status).toBe(200); + }) + )); + it('keeps browser app state away from server-owned keys', () => withTempStorage((storage) => withServer(app(storage), async (baseUrl) => { diff --git a/apps/web-backend/src/app/storage/user-state-router.ts b/apps/web-backend/src/app/storage/user-state-router.ts index b5a4122b0..bc5f7b8f0 100644 --- a/apps/web-backend/src/app/storage/user-state-router.ts +++ b/apps/web-backend/src/app/storage/user-state-router.ts @@ -1,8 +1,9 @@ -import express, { Router } from 'express'; +import express, { Router, type Request, type Response } from 'express'; import { isServerUserStateNamespace, type ServerUserStateNamespace, } from '@iptvnator/shared/interfaces'; +import type { ExpectedTag } from './server-user-state-repository'; import type { ServerStorage } from './server-storage'; import { StorageRequestError } from './storage-route-errors'; @@ -18,11 +19,30 @@ export function createUserStateRouter(storage: ServerStorage): Router { const json = express.json({ limit: USER_STATE_BODY_LIMIT }); router.get('/user-state', (_req, res) => { - res.set('Cache-Control', 'no-store').json({ - entries: storage.userState.all(), - }); + res.set('Cache-Control', 'no-store').json( + storage.userState.allWithTags() + ); }); + const write = (req: Request, res: Response, value: unknown) => { + const result = storage.userState.writeIf( + requireNamespace(String(req.params['namespace'])), + String(req.params['playlistId']), + value, + expectedTag(req) + ); + if ('conflict' in result) { + res.status(412).json({ + message: 'The entry changed since it was read.', + status: 412, + code: 'conflict', + current: result.conflict, + }); + return; + } + res.json({ success: true, tag: result.tag }); + }; + router.put('/user-state/:namespace/:playlistId', json, (req, res) => { const body = req.body as { value?: unknown } | undefined; if (!body || typeof body !== 'object' || !('value' in body)) { @@ -32,20 +52,11 @@ export function createUserStateRouter(storage: ServerStorage): Router { 'Expected { value }.' ); } - storage.userState.put( - requireNamespace(req.params.namespace), - req.params.playlistId, - body.value - ); - res.json({ success: true }); + write(req, res, body.value ?? null); }); router.delete('/user-state/:namespace/:playlistId', (req, res) => { - storage.userState.delete( - requireNamespace(req.params.namespace), - req.params.playlistId - ); - res.json({ success: true }); + write(req, res, undefined); }); router.get('/app-state/:key', (req, res) => { @@ -75,6 +86,18 @@ export function createUserStateRouter(storage: ServerStorage): Router { return router; } +/** + * `If-Match: ""` expects that tag, `If-None-Match: *` expects no + * entry; without either the write is unconditional, as older clients send. + */ +function expectedTag(req: Request): ExpectedTag { + const ifMatch = req.get('if-match')?.trim(); + if (ifMatch) { + return ifMatch.replace(/^W\//, '').replace(/^"|"$/g, ''); + } + return req.get('if-none-match')?.trim() === '*' ? null : undefined; +} + function requireNamespace(value: string): ServerUserStateNamespace { if (!isServerUserStateNamespace(value)) { throw new StorageRequestError( 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 ea6fd8251..2f9a15fa4 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 @@ -11,7 +11,10 @@ import { 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 { + isBackendUrl, + ServerAccessTokenInterceptor, +} from './server-access-token.interceptor'; import { SERVER_STORAGE_CONFIG } from './server-storage.config'; describe('ServerAccessTokenInterceptor', () => { @@ -92,3 +95,36 @@ describe('ServerAccessTokenInterceptor', () => { ).toBeNull(); }); }); + +describe('isBackendUrl', () => { + const page = 'http://nas.local:4333/workspace'; + + it('matches the backend path and nothing else', () => { + expect(isBackendUrl('/api/xtream', '/api', page)).toBe(true); + expect(isBackendUrl('/api', '/api', page)).toBe(true); + expect(isBackendUrl('/apiary', '/api', page)).toBe(false); + expect(isBackendUrl('https://other.test/api/x', '/api', page)).toBe( + false + ); + }); + + it('keeps other origins out for a root deployment', () => { + expect(isBackendUrl('/v1/playlists', '/', page)).toBe(true); + expect( + isBackendUrl('https://api.github.com/repos/x/releases', '/', page) + ).toBe(false); + }); + + it('handles an absolute backend URL', () => { + expect( + isBackendUrl( + 'http://nas.local:3000/parse', + 'http://nas.local:3000', + page + ) + ).toBe(true); + expect(isBackendUrl('/parse', 'http://nas.local:3000', page)).toBe( + false + ); + }); +}); 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 609814ba7..afd0f2ed2 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 @@ -22,16 +22,13 @@ import { SERVER_STORAGE_CONFIG } from './server-storage.config'; @Injectable() export class ServerAccessTokenInterceptor implements HttpInterceptor { private readonly status = inject(ServerStorageStatusService); - private readonly baseUrl = inject(SERVER_STORAGE_CONFIG).baseUrl.replace( - /\/+$/, - '' - ); + private readonly baseUrl = inject(SERVER_STORAGE_CONFIG).baseUrl; intercept( request: HttpRequest, next: HttpHandler ): Observable> { - if (!request.url.startsWith(this.baseUrl)) { + if (!isBackendUrl(request.url, this.baseUrl)) { return next.handle(request); } const token = readServerAccessToken(); @@ -57,3 +54,29 @@ export class ServerAccessTokenInterceptor implements HttpInterceptor { ); } } + +/** + * Whether `url` goes to the backend: the same origin as the backend URL and + * a path inside it. A plain prefix check fails for `BACKEND_URL=/`, where + * every URL, other origins included, would carry the token. + */ +export function isBackendUrl( + url: string, + baseUrl: string, + pageUrl = globalThis.location?.href ?? 'http://localhost/' +): boolean { + try { + const base = new URL(baseUrl, pageUrl); + const target = new URL(url, pageUrl); + const root = base.pathname.endsWith('/') + ? base.pathname + : `${base.pathname}/`; + return ( + target.origin === base.origin && + (target.pathname === base.pathname || + target.pathname.startsWith(root)) + ); + } catch { + return false; + } +} 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 1c3b49f40..c3cae2b28 100644 --- a/libs/services/src/lib/server-storage/server-api-client.ts +++ b/libs/services/src/lib/server-storage/server-api-client.ts @@ -1,4 +1,7 @@ -import type { ServerStorageErrorBody } from '@iptvnator/shared/interfaces'; +import type { + ServerStorageErrorBody, + ServerUserStateCurrent, +} from '@iptvnator/shared/interfaces'; import { clearServerAccessToken, serverAuthorizationHeaders, @@ -14,7 +17,9 @@ export class ServerApiError extends Error { constructor( readonly status: number, readonly code: string | undefined, - message: string + message: string, + /** A `412` conflict's user-state entry as the server has it. */ + readonly current?: ServerUserStateCurrent ) { super(message); this.name = 'ServerApiError'; @@ -91,7 +96,8 @@ export class ServerApiClient { const error = new ServerApiError( response.status, body.code, - body.message + body.message, + body.current ); if ( isWrite && @@ -133,16 +139,16 @@ export class ServerApiClient { sendJson( method: 'PUT' | 'POST', path: string, - body: unknown + body: unknown, + headers: Record = {} ): Promise { return this.request(path, { method, body: JSON.stringify(body), - headers: { 'Content-Type': 'application/json' }, + headers: { 'Content-Type': 'application/json', ...headers }, }); } - /** Best effort while the page is hidden or closed; errors are ignored. */ /** * A write that survives the page. Resolves whether the server took it; * browsers reject keepalive bodies above 64 KiB, which resolves false. @@ -150,7 +156,8 @@ export class ServerApiClient { async sendKeepalive( method: 'PUT' | 'DELETE', path: string, - body?: unknown + body?: unknown, + headers: Record = {} ): Promise { try { const response = await this.fetchFn()(this.url(path), { @@ -158,6 +165,7 @@ export class ServerApiClient { keepalive: true, headers: { ...serverAuthorizationHeaders(), + ...headers, ...(body === undefined ? {} : { 'Content-Type': 'application/json' }), @@ -183,12 +191,13 @@ export class ServerApiClient { async function readErrorBody( response: Response -): Promise<{ message: string; code?: string }> { +): Promise> { try { const body = (await response.json()) as Partial; return { message: body.message ?? response.statusText, code: body.code, + current: body.current, }; } catch { return { message: response.statusText }; 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 56b316aec..f56b7282c 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 @@ -123,7 +123,8 @@ describe('ServerMigrationService', () => { expect(session.api.sendJson).toHaveBeenCalledWith( 'PUT', '/user-state/xtream-favorites/x', - { value: ['live:1'] } + { value: ['live:1'] }, + {} ); expect(session.api.sendJson).toHaveBeenCalledTimes(1); // Queued edits were sent before the combined values were read. @@ -175,11 +176,13 @@ describe('ServerMigrationService', () => { 'PUT', '/user-state/xtream-favorites/server-x', { value: ['live:1'] }, + {}, ], [ 'PUT', '/user-state/xtream-recent-items/server-x', { value: ['vod:9', 'vod:1'] }, + {}, ], ]); }); @@ -209,6 +212,7 @@ describe('ServerMigrationService', () => { 'PUT', '/user-state/xtream-favorites/x', { value: ['live:1', 'live:2'] }, + {}, ], ]); }); 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 e1d334c5b..4791ed4ef 100644 --- a/libs/services/src/lib/server-storage/server-migration.service.ts +++ b/libs/services/src/lib/server-storage/server-migration.service.ts @@ -14,6 +14,7 @@ import { unionItems, } from './server-migration-merge'; import { planServerMigration } from './server-migration-plan'; +import { preconditionHeaders } from './server-user-data-entries'; import { typeLegacyXtreamRefs, type XtreamUserData, @@ -202,7 +203,7 @@ export class ServerMigrationService { if (targets.size === 0) { return 0; } - const { entries } = + const { entries, tags } = await this.session.api.getJson( '/user-state' ); @@ -250,10 +251,17 @@ export class ServerMigrationService { continue; } try { + // On the version read above: a device that changed it + // meanwhile wins, and the move stays on offer. await this.session.api.sendJson( 'PUT', `/user-state/${namespace}/${encodeURIComponent(serverId)}`, - { value } + { value }, + tags + ? preconditionHeaders( + tags[namespace]?.[serverId] ?? null + ) + : {} ); } catch { failed += 1; 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 bdeb43bc0..8a3c034b2 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 @@ -141,4 +141,31 @@ describe('ServerPlaylistDocumentStore', () => { await expect(readThenEdit('"b.0.4"')).resolves.toEqual({}); }); }); + + it('clears user-data writes before deleting on the server', async () => { + const order: string[] = []; + const api = { + request: jest.fn(async (path: string) => { + order.push(`DELETE ${path}`); + return fakeResponse(200, { success: true }); + }), + getJson: jest.fn(), + readJson: jest.fn(), + sendJson: jest.fn(), + }; + const store = new ServerPlaylistDocumentStore( + api as never, + async (playlistId) => void order.push(`discard ${playlistId}`) + ); + + await store.dbDeletePlaylist('p1'); + await store.dbDeleteAllPlaylists(); + + expect(order).toEqual([ + 'discard p1', + 'DELETE /playlists/p1', + 'discard undefined', + 'DELETE /playlists', + ]); + }); }); 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 d7691bef1..076260ec1 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 @@ -30,7 +30,16 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { /** Served payload object → the ETag of the document it came from. */ private readonly servedPayloads = new WeakMap(); - constructor(private readonly api: ServerApi) {} + constructor( + private readonly api: ServerApi, + /** + * Runs before a playlist (or, without an id, every playlist) is + * deleted on the server, so no user-data write lands afterwards. + */ + private readonly beforeDelete: ( + playlistId?: string + ) => Promise = async () => undefined + ) {} dbGetAppPlaylists(): Promise { return this.api.getJson('/playlists'); @@ -111,6 +120,7 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { } async dbDeletePlaylist(playlistId: string): Promise { + await this.beforeDelete(playlistId); await this.api.request(`/playlists/${encodeURIComponent(playlistId)}`, { method: 'DELETE', }); @@ -119,6 +129,7 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore { } async dbDeleteAllPlaylists(): Promise { + await this.beforeDelete(); await this.api.request('/playlists', { method: 'DELETE' }); this.cache.clear(); return { success: true }; 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 645279984..b741bfb29 100644 --- a/libs/services/src/lib/server-storage/server-storage-session.ts +++ b/libs/services/src/lib/server-storage/server-storage-session.ts @@ -40,7 +40,10 @@ export class ServerStorageSession { status: this.status, onWriteFailure: (error) => this.announceWriteFailure(error), }); - readonly documentStore = new ServerPlaylistDocumentStore(this.api); + readonly documentStore = new ServerPlaylistDocumentStore( + this.api, + (playlistId) => this.userData.discard(playlistId) + ); readonly userData = new ServerUserDataStorage( this.api, // Lazy: a browser that denies storage must not stop the app here. diff --git a/libs/services/src/lib/server-storage/server-unsent-writes.ts b/libs/services/src/lib/server-storage/server-unsent-writes.ts index 7caadc98c..7c76d5d15 100644 --- a/libs/services/src/lib/server-storage/server-unsent-writes.ts +++ b/libs/services/src/lib/server-storage/server-unsent-writes.ts @@ -61,6 +61,10 @@ export interface UnsentWrite { readonly playlistId: string; /** `undefined` deletes the entry. */ readonly value: unknown; + /** The entry the edit was made against, to replay it on a conflict. */ + readonly before?: unknown; + /** The server version it is based on (see `PendingEntry.tag`). */ + readonly tag?: string | null; } interface StoredWrite { @@ -68,6 +72,8 @@ interface StoredWrite { readonly playlistId: string; readonly value?: unknown; readonly remove?: true; + readonly before?: unknown; + readonly tag?: string | null; } export function readUnsentWrites( @@ -93,6 +99,8 @@ export function readUnsentWrites( namespace: write.namespace, playlistId: write.playlistId, value: write.remove ? undefined : write.value, + before: write.before, + tag: write.tag, }); } } @@ -110,10 +118,13 @@ export function saveUnsentWrites( return; } const stored: StoredWrite[] = writes.map( - ({ namespace, playlistId, value }) => - value === undefined - ? { namespace, playlistId, remove: true } - : { namespace, playlistId, value } + ({ namespace, playlistId, value, before, tag }) => ({ + namespace, + playlistId, + ...(value === undefined ? { remove: true } : { value }), + ...(before === undefined ? {} : { before }), + ...(tag === undefined ? {} : { tag }), + }) ); local.setItem(writesKey(instanceId), JSON.stringify(stored)); } catch { diff --git a/libs/services/src/lib/server-storage/server-user-data-entries.ts b/libs/services/src/lib/server-storage/server-user-data-entries.ts index 84ca4fd67..51decac5c 100644 --- a/libs/services/src/lib/server-storage/server-user-data-entries.ts +++ b/libs/services/src/lib/server-storage/server-user-data-entries.ts @@ -8,6 +8,23 @@ export interface PendingEntry { readonly value: unknown; /** The edit this value came from; a retry never replaces a newer one. */ readonly edit: number; + /** The entry as the edit found it, to replay the edit on a conflict. */ + readonly before: unknown; + /** + * The server version the edit is based on: a tag, `null` for an entry + * seen absent, `undefined` when unknown (an unconditional write). + */ + readonly tag?: string | null; +} + +/** `If-Match` / `If-None-Match` for a write based on `tag`. */ +export function preconditionHeaders( + tag: string | null | undefined +): Record { + if (tag === undefined) { + return {}; + } + return tag === null ? { 'If-None-Match': '*' } : { 'If-Match': `"${tag}"` }; } export function entryKey(entry: PendingEntry): string { diff --git a/libs/services/src/lib/server-storage/server-user-data-queue.ts b/libs/services/src/lib/server-storage/server-user-data-queue.ts new file mode 100644 index 000000000..c346dfff1 --- /dev/null +++ b/libs/services/src/lib/server-storage/server-user-data-queue.ts @@ -0,0 +1,404 @@ +import type { + ServerUserStateCurrent, + ServerUserStateNamespace, + ServerUserStateWriteResponse, +} from '@iptvnator/shared/interfaces'; +import { ServerApiError, type ServerApiClient } from './server-api-client'; +import { + entryKey, + entryPath, + type PendingEntry, + preconditionHeaders, +} from './server-user-data-entries'; +import { applyEdit } from './server-user-data-edit'; +import { newestWrites } from './server-unsent-writes'; + +const FLUSH_DELAY_MS = 500; + +type ServerApi = Pick< + ServerApiClient, + 'readJson' | 'sendJson' | 'request' | 'sendKeepalive' +>; + +export interface UserDataQueueHooks { + /** The queue or the writes travelling changed: persist them. */ + readonly changed: () => void; + /** A conflict replay changed what an entry should show. */ + readonly show: ( + namespace: ServerUserStateNamespace, + playlistId: string, + value: unknown + ) => void; + readonly writeFailed: (error: unknown) => void; +} + +/** + * The writes of `ServerUserDataStorage`: one entry per playlist and + * namespace, sent in batches one after another, each based on the server + * version (tag) it was read at. A `412` replays the edit (before → after) + * on the server's current value; a failure keeps it queued unless a newer + * edit replaced it. + */ +export class UserDataWriteQueue { + /** Server version tag per entry; absent until the first read. */ + private readonly tags = new Map< + ServerUserStateNamespace, + Record + >(); + private readonly pending = new Map(); + /** Writes still travelling (normal or page-hide), by entry key. */ + private readonly inFlight = 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(); + private writesInFlight = 0; + /** Edit counter per `namespace|playlistId`, for re-reads to keep them. */ + private readonly edits = new Map(); + private editCount = 0; + + constructor( + private readonly api: ServerApi, + private readonly hooks: UserDataQueueHooks + ) {} + + get lastEdit(): number { + return this.editCount; + } + + hasUnsent(): boolean { + return this.pending.size > 0 || this.inFlight.size > 0; + } + + /** What the server has not confirmed, the newest write per entry. */ + unsentWrites(): PendingEntry[] { + return newestWrites([ + ...this.inFlight.values(), + ...this.pending.values(), + ]); + } + + unconfirmedKeys(): string[] { + return [...this.pending.keys(), ...this.inFlight.keys()]; + } + + /** Entries a re-read must not replace: unconfirmed or edited since. */ + unsavedSince(edit: number): Set { + const keys = new Set(this.unconfirmedKeys()); + for (const [key, made] of this.edits) { + if (made > edit) { + keys.add(key); + } + } + return keys; + } + + /** Forgets edit numbers a re-read has covered. */ + settledUpTo(edit: number): void { + for (const [key, made] of this.edits) { + if ( + made <= edit && + !this.pending.has(key) && + !this.inFlight.has(key) + ) { + this.edits.delete(key); + } + } + } + + setTags( + namespace: ServerUserStateNamespace, + tags: Readonly> + ): void { + this.tags.set(namespace, { ...tags }); + } + + /** + * A newer edit of an entry that is still queued or travelling keeps + * that write's baseline and tag, so a conflict replays both edits. + */ + queue( + namespace: ServerUserStateNamespace, + playlistId: string, + value: unknown, + before: unknown, + tag = this.tagFor(namespace, playlistId) + ): void { + const key = `${namespace}|${playlistId}`; + const earlier = this.pending.get(key) ?? this.inFlight.get(key); + const entry: PendingEntry = { + namespace, + playlistId, + value, + edit: ++this.editCount, + before: earlier ? earlier.before : before, + tag: earlier ? earlier.tag : tag, + }; + this.pending.set(key, entry); + this.edits.set(key, entry.edit); + } + + scheduleFlush(): void { + if (this.pending.size === 0) { + return; + } + // Saved at once: another tab's re-read must not replace them. + this.hooks.changed(); + if (this.flushTimer !== null) { + return; + } + this.flushTimer = setTimeout(() => { + this.flushTimer = null; + void this.flush(); + }, FLUSH_DELAY_MS); + } + + /** Sends the queued entries after any batch still travelling. */ + flush(): Promise { + const run = this.sending.then(async () => { + const batch = this.takePending(); + if (batch.length === 0) { + return; + } + for (const entry of batch) { + this.inFlight.set(entryKey(entry), entry); + } + this.writesInFlight += 1; + try { + await Promise.all(batch.map((entry) => this.send(entry))); + } finally { + this.writesInFlight -= 1; + } + }); + this.sending = run.then( + () => undefined, + () => undefined + ); + return this.sending; + } + + /** + * Page hide: `keepalive` requests survive the page. They go out at once + * unless a write is still travelling, which they must not overtake, and + * later writes wait for them in turn. + */ + flushWithKeepalive(): void { + const entries = this.takePending(); + if (entries.length === 0) { + // Writes still travelling may be cut off with the page. + this.hooks.changed(); + return; + } + const sendAll = async (): Promise => { + this.writesInFlight += 1; + try { + await Promise.all( + entries.map((entry) => this.sendKeepalive(entry)) + ); + } finally { + this.writesInFlight -= 1; + } + }; + for (const entry of entries) { + this.inFlight.set(entryKey(entry), entry); + } + this.hooks.changed(); + // Idle: send synchronously, while the hiding page still runs. + const run = + this.writesInFlight === 0 ? sendAll() : this.sending.then(sendAll); + this.sending = run.then( + () => undefined, + () => undefined + ); + } + + /** + * Drops queued edits whose key `matches` and forgets their edit numbers, + * so a travelling write that fails is not retried; resolves once the + * travelling writes have landed. Without a playlist id, tags go too. + */ + async discard( + matches: (key: string) => boolean, + playlistId?: string + ): Promise { + for (const store of [this.pending, this.edits]) { + for (const key of [...store.keys()]) { + if (matches(key)) { + store.delete(key); + } + } + } + await this.sending; + for (const [namespace, tags] of this.tags) { + if (playlistId === undefined) { + this.tags.set(namespace, {}); + } else { + delete tags[playlistId]; + } + } + this.hooks.changed(); + } + + /** `undefined` until the server's tags are known (or not offered). */ + private tagFor( + namespace: ServerUserStateNamespace, + playlistId: string + ): string | null | undefined { + const tags = this.tags.get(namespace); + return tags ? (tags[playlistId] ?? null) : undefined; + } + + private setTag( + namespace: ServerUserStateNamespace, + playlistId: string, + tag: string | null + ): void { + const tags = this.tags.get(namespace); + if (!tags) { + return; + } + if (tag === null) { + delete tags[playlistId]; + } else { + tags[playlistId] = tag; + } + } + + private takePending(): PendingEntry[] { + if (this.flushTimer !== null) { + clearTimeout(this.flushTimer); + this.flushTimer = null; + } + const batch = [...this.pending.values()]; + this.pending.clear(); + 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): void { + const key = entryKey(entry); + if (this.edits.get(key) === entry.edit && !this.pending.has(key)) { + this.pending.set(key, entry); + } + } + + /** A newer write of the same entry may have replaced it meanwhile. */ + private retireInFlight(entry: PendingEntry): void { + const key = entryKey(entry); + if (this.inFlight.get(key)?.edit === entry.edit) { + this.inFlight.delete(key); + } + } + + private async sendKeepalive(entry: PendingEntry): Promise { + const path = entryPath(entry); + const headers = preconditionHeaders(entry.tag); + const ok = await (entry.value === undefined + ? this.api.sendKeepalive('DELETE', path, undefined, headers) + : this.api.sendKeepalive( + 'PUT', + path, + { value: entry.value }, + headers + )); + this.retireInFlight(entry); + if (!ok) { + this.requeue(entry); + } + this.hooks.changed(); + } + + private async send(entry: PendingEntry): Promise { + const headers = preconditionHeaders(entry.tag); + try { + const response = + entry.value === undefined + ? await this.api.request(entryPath(entry), { + method: 'DELETE', + headers, + }) + : await this.api.sendJson( + 'PUT', + entryPath(entry), + { value: entry.value }, + headers + ); + const { tag } = await this.api + .readJson>(response) + .catch(() => ({ tag: undefined })); + this.retireInFlight(entry); + if (tag !== undefined) { + this.confirmed(entry, tag); + } + } catch (error) { + this.retireInFlight(entry); + if ( + error instanceof ServerApiError && + error.status === 412 && + error.current + ) { + this.replayOnConflict(entry, error.current); + } else { + this.requeue(entry); + this.hooks.writeFailed(error); + } + } + this.hooks.changed(); + } + + /** The server took `entry`: a newer queued edit now builds on it. */ + private confirmed(entry: PendingEntry, tag: string | null): void { + this.setTag(entry.namespace, entry.playlistId, tag); + const key = entryKey(entry); + const next = this.pending.get(key); + if (next && next.before === entry.before && next.tag === entry.tag) { + this.pending.set(key, { ...next, before: entry.value, tag }); + } + } + + /** + * Another device changed the entry: apply this edit (before → after) to + * the server's current value, show the result and send it on that + * version, unless a newer edit of the entry will. + */ + private replayOnConflict( + entry: PendingEntry, + current: ServerUserStateCurrent + ): void { + this.setTag(entry.namespace, entry.playlistId, current.tag); + const key = entryKey(entry); + const next = this.pending.get(key); + if (next) { + // The newer edit kept this one's baseline: replay both at once. + this.pending.set(key, { + ...next, + value: applyEdit(current.value, next.before, next.value), + before: current.value, + tag: current.tag, + }); + } else if (this.edits.get(key) === entry.edit) { + const merged = applyEdit(current.value, entry.before, entry.value); + if (JSON.stringify(merged) !== JSON.stringify(current.value)) { + this.pending.set(key, { + ...entry, + value: merged, + before: current.value, + tag: current.tag, + }); + } else { + this.hooks.show(entry.namespace, entry.playlistId, merged); + return; + } + } else { + return; + } + this.hooks.show( + entry.namespace, + entry.playlistId, + this.pending.get(key)?.value + ); + this.scheduleFlush(); + } +} 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 cece20be0..9f6aa295c 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 @@ -35,6 +35,7 @@ describe('ServerUserDataStorage', () => { sendJson: jest.fn().mockResolvedValue(undefined), request: jest.fn().mockResolvedValue(undefined), sendKeepalive: jest.fn().mockResolvedValue(true), + readJson: jest.fn().mockResolvedValue({}), }; const onWriteFailure = jest.fn(); const storage = new ServerUserDataStorage( @@ -67,11 +68,12 @@ describe('ServerUserDataStorage', () => { expect(api.sendJson).toHaveBeenCalledWith( 'PUT', '/user-state/xtream-favorites/p2', - { value: ['live:9'] } + { value: ['live:9'] }, + {} ); expect(api.request).toHaveBeenCalledWith( '/user-state/xtream-favorites/p1', - { method: 'DELETE' } + { method: 'DELETE', headers: {} } ); expect(api.sendJson).toHaveBeenCalledTimes(1); }); @@ -95,7 +97,8 @@ describe('ServerUserDataStorage', () => { expect(api.sendKeepalive).toHaveBeenCalledWith( 'PUT', '/user-state/xtream-recent-items/p1', - { value: [] } + { value: [] }, + {} ); await jest.runOnlyPendingTimersAsync(); expect(api.sendJson).not.toHaveBeenCalled(); @@ -138,7 +141,8 @@ describe('ServerUserDataStorage', () => { expect(api.sendJson).toHaveBeenCalledWith( 'PUT', '/user-state/xtream-favorites/p2', - { value: ['live:2'] } + { value: ['live:2'] }, + {} ); // The next edit does not send a delete for the kept entry. @@ -149,7 +153,7 @@ describe('ServerUserDataStorage', () => { await jest.runOnlyPendingTimersAsync(); expect(api.request).not.toHaveBeenCalledWith( '/user-state/xtream-favorites/p2', - { method: 'DELETE' } + { method: 'DELETE', headers: {} } ); }); @@ -212,7 +216,8 @@ describe('ServerUserDataStorage', () => { expect(api.sendJson).toHaveBeenLastCalledWith( 'PUT', '/user-state/xtream-favorites/p1', - { value: ['live:1', 'live:2'] } + { value: ['live:1', 'live:2'] }, + {} ); expect(JSON.parse(storage.getItem('xtream-favorites') ?? '')).toEqual({ p1: ['live:1', 'live:2'], @@ -256,7 +261,8 @@ describe('ServerUserDataStorage', () => { expect(api.sendKeepalive).toHaveBeenCalledWith( 'PUT', '/user-state/xtream-favorites/p1', - { value: ['b'] } + { value: ['b'] }, + {} ); }); @@ -277,7 +283,8 @@ describe('ServerUserDataStorage', () => { expect(api.sendJson).toHaveBeenCalledWith( 'PUT', '/user-state/xtream-favorites/p1', - { value: ['big'] } + { value: ['big'] }, + {} ); }); @@ -321,7 +328,8 @@ describe('ServerUserDataStorage', () => { expect(api.sendJson).toHaveBeenCalledWith( 'PUT', '/user-state/xtream-favorites/p1', - { value: ['new'] } + { value: ['new'] }, + {} ); }); @@ -354,6 +362,7 @@ describe('ServerUserDataStorage', () => { 'PUT', '/user-state/xtream-favorites/p1', { value: ['live:1', 'a', 'c'] }, + {}, ], ]); expect(JSON.parse(storage.getItem('xtream-favorites') ?? '')).toEqual({ @@ -380,7 +389,8 @@ describe('ServerUserDataStorage', () => { expect(api.sendJson).toHaveBeenCalledWith( 'PUT', '/user-state/xtream-favorites/p1', - { value: ['live:1', 'live:2', 'live:9'] } + { value: ['live:1', 'live:2', 'live:9'] }, + {} ); }); @@ -442,6 +452,7 @@ describe('ServerUserDataStorage', () => { 'PUT', '/user-state/xtream-favorites/p1', { value: ['live:1', 'live:3'] }, + {}, ], ]); }); @@ -457,16 +468,18 @@ describe('ServerUserDataStorage', () => { await first.storage.flush(); expect(first.storage.hasUnsentWrites()).toBe(true); expect( - first.local.values.get('iptvnator-server:unsent-writes:default') - ).toBe( - JSON.stringify([ - { - namespace: 'xtream-favorites', - playlistId: 'p1', - value: ['live:1', 'live:9'], - }, - ]) - ); + JSON.parse( + first.local.values.get( + 'iptvnator-server:unsent-writes:default' + ) ?? '[]' + ) + ).toMatchObject([ + { + namespace: 'xtream-favorites', + playlistId: 'p1', + value: ['live:1', 'live:9'], + }, + ]); // Reload: a new instance over the same browser storage. const api = { @@ -486,7 +499,8 @@ describe('ServerUserDataStorage', () => { expect(api.sendJson).toHaveBeenCalledWith( 'PUT', '/user-state/xtream-favorites/p1', - { value: ['live:1', 'live:9'] } + { value: ['live:1', 'live:9'] }, + {} ); expect(reloaded.hasUnsentWrites()).toBe(false); expect( @@ -509,7 +523,7 @@ describe('ServerUserDataStorage', () => { local.values.get('iptvnator-server:unsent-writes:default') ?? '[]' ) - ).toEqual([ + ).toMatchObject([ { namespace: 'xtream-favorites', playlistId: 'p1', value: ['x'] }, ]); }); @@ -570,7 +584,7 @@ describe('ServerUserDataStorage', () => { local.values.get('iptvnator-server:unsent-writes:default') ?? '[]' ) - ).toEqual([ + ).toMatchObject([ { namespace: 'xtream-favorites', playlistId: 'p1', value: ['new'] }, ]); }); @@ -592,7 +606,8 @@ describe('ServerUserDataStorage', () => { expect(api.sendJson).toHaveBeenCalledWith( 'PUT', '/user-state/xtream-favorites/p1', - { value: ['S', 'A', 'B'] } + { value: ['S', 'A', 'B'] }, + {} ); }); @@ -623,7 +638,8 @@ describe('ServerUserDataStorage', () => { expect(api.sendJson).toHaveBeenCalledWith( 'PUT', '/user-state/xtream-favorites/p1', - { value: ['theirs', 'mine'] } + { value: ['theirs', 'mine'] }, + {} ); }); @@ -636,6 +652,7 @@ describe('ServerUserDataStorage', () => { sendJson: jest.fn().mockResolvedValue(undefined), request: jest.fn().mockResolvedValue(undefined), sendKeepalive: jest.fn().mockResolvedValue(true), + readJson: jest.fn().mockResolvedValue({}), }; const tab = () => new ServerUserDataStorage( @@ -691,7 +708,8 @@ describe('ServerUserDataStorage', () => { expect(api.sendJson).toHaveBeenLastCalledWith( 'PUT', '/user-state/xtream-favorites/p1', - { value: ['S', 'B', 'C'] } + { value: ['S', 'B', 'C'] }, + {} ); }); }); 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 92305baf3..2973f94e7 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 @@ -6,18 +6,12 @@ import { } from '@iptvnator/shared/interfaces'; import type { UserDataStorage } from '../user-data-storage'; import type { ServerApiClient } from './server-api-client'; -import { - entryKey, - entryPath, - parseMap, - type PendingEntry, - splitEditKey, -} from './server-user-data-entries'; +import { parseMap, splitEditKey } from './server-user-data-entries'; import { type DeferredEdit, refreshedNamespace } from './server-user-data-edit'; +import { UserDataWriteQueue } from './server-user-data-queue'; import { claimCache, markUnconfirmed, - newestWrites, othersUnconfirmed, readDeferredEdits, readUnsentWrites, @@ -27,31 +21,22 @@ import { /** Server-mode cache keys; the legacy unprefixed keys are never touched. */ export const SERVER_CACHE_PREFIX = 'iptvnator-server:'; -const FLUSH_DELAY_MS = 500; type ServerApi = Pick< ServerApiClient, - 'getJson' | 'sendJson' | 'request' | 'sendKeepalive' + 'getJson' | 'readJson' | 'sendJson' | 'request' | 'sendKeepalive' >; /** * 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 use plain storage. + * maps keyed by playlist id; a write compares the new map with the one the + * caller read and sends only the playlist entries that changed, each based + * on the server version it was read at. When another device changed the + * entry meanwhile, the server answers `412` and the edit is replayed on its + * current value (`UserDataWriteQueue`). Other keys use plain storage. */ export class ServerUserDataStorage implements UserDataStorage { - private readonly synced = new Map< - ServerUserStateNamespace, - Record - >(); - 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(); - private writesInFlight = 0; - /** Writes still travelling (normal or page-hide), by entry key. */ - private readonly inFlight = new Map(); + private readonly writes: UserDataWriteQueue; private unsentSaved = false; /** * Edits made before the first successful read, as before → after: the @@ -66,14 +51,11 @@ export class ServerUserDataStorage implements UserDataStorage { private readonly overflow = new Map(); /** 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; constructor( private readonly api: ServerApi, private readonly local: UserDataStorage, - private readonly onWriteFailure: (error: unknown) => void, + onWriteFailure: (error: unknown) => void, /** Scopes the cache and saved writes to one server database. */ private readonly instanceId = 'default', /** @@ -83,13 +65,25 @@ export class ServerUserDataStorage implements UserDataStorage { */ private readonly tabStore: UserDataStorage = local ) { + this.writes = new UserDataWriteQueue(api, { + changed: () => this.saveUnsent(), + show: (namespace, playlistId, value) => + this.showEntry(namespace, playlistId, value), + writeFailed: onWriteFailure, + }); claimCache(local, instanceId); // Writes a previous visit could not confirm go out with the first // flush, and the first read keeps them. for (const write of readUnsentWrites(tabStore, instanceId)) { - this.queue(write.namespace, write.playlistId, write.value); + this.writes.queue( + write.namespace, + write.playlistId, + write.value, + write.before, + write.tag + ); } - this.unsentSaved = this.pending.size > 0; + this.unsentSaved = this.writes.hasUnsent(); for (const edit of readDeferredEdits(tabStore, instanceId)) { this.deferred.set(`${edit.namespace}|${edit.playlistId}`, edit); } @@ -97,11 +91,7 @@ export class ServerUserDataStorage implements UserDataStorage { /** Whether a write is queued, being retried, or still travelling. */ hasUnsentWrites(): boolean { - return ( - this.pending.size > 0 || - this.inFlight.size > 0 || - this.deferred.size > 0 - ); + return this.writes.hasUnsent() || this.deferred.size > 0; } getItem(key: string): string | null { @@ -144,21 +134,51 @@ export class ServerUserDataStorage implements UserDataStorage { 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(); - const { entries } = - await this.api.getJson('/user-state'); - const unsaved = new Set([ - ...this.pending.keys(), - ...this.inFlight.keys(), - ]); - for (const [key, edit] of this.edits) { - if (edit > editsBefore) { - unsaved.add(key); + /** Sends the queued entries after any batch still travelling. */ + flush(): Promise { + return this.writes.flush(); + } + + /** Page hide: sends the queue with requests that survive the page. */ + flushWithKeepalive(): void { + this.writes.flushWithKeepalive(); + } + + /** + * Before a playlist (or every playlist) is deleted on the server: drop + * its queued and early edits and its cache, and let writes already + * travelling land first, so none recreates deleted data afterwards. + */ + async discard(playlistId?: string): Promise { + const matches = (key: string) => + playlistId === undefined || splitEditKey(key)[1] === playlistId; + for (const key of [...this.deferred.keys()]) { + if (matches(key)) { + this.deferred.delete(key); } } + saveDeferredEdits(this.tabStore, this.instanceId, [ + ...this.deferred.values(), + ]); + await this.writes.discard(matches, playlistId); + for (const namespace of SERVER_USER_STATE_NAMESPACES) { + const map = this.baseline(namespace); + if (playlistId === undefined) { + this.writeCache(namespace, null); + } else if (playlistId in map) { + delete map[playlistId]; + this.writeCache(namespace, JSON.stringify(map)); + } + } + } + + private async hydrateNow(): Promise { + // Edits from here on may miss the read: queued, or still travelling. + const editsBefore = this.writes.lastEdit; + await this.writes.flush(); + const { entries, tags } = + await this.api.getJson('/user-state'); + const unsaved = this.writes.unsavedSince(editsBefore); // Another tab's unconfirmed writes are in the shared cache too. for (const key of othersUnconfirmed( this.local, @@ -179,7 +199,7 @@ export class ServerUserDataStorage implements UserDataStorage { } const value = refreshedNamespace( entries[namespace] ?? {}, - parseMap(this.readCache(namespace) ?? '{}'), + this.baseline(namespace), keepCached, deferred.filter((edit) => edit.namespace === namespace) ); @@ -187,92 +207,28 @@ export class ServerUserDataStorage implements UserDataStorage { // A value too big for browser storage stays readable in memory, // so callers and the diff baseline see the same map. this.writeCache(namespace, JSON.stringify(value)); - this.synced.set(namespace, value); - } - for (const [key, edit] of this.edits) { - if ( - edit <= editsBefore && - !this.pending.has(key) && - !this.inFlight.has(key) - ) { - this.edits.delete(key); + if (tags) { + this.writes.setTags(namespace, tags[namespace] ?? {}); } } + this.writes.settledUpTo(editsBefore); if (!this.hydrated) { // From here edits go straight to the queue. this.hydrated = true; this.deferred.clear(); for (const edit of deferred) { - this.queue( + this.writes.queue( edit.namespace, edit.playlistId, - refreshed.get(edit.namespace)?.[edit.playlistId] + refreshed.get(edit.namespace)?.[edit.playlistId], + entries[edit.namespace]?.[edit.playlistId] ); } saveDeferredEdits(this.tabStore, this.instanceId, []); - this.scheduleFlush(); + this.writes.scheduleFlush(); } } - /** Sends the queued entries after any batch still travelling. */ - flush(): Promise { - const run = this.sending.then(async () => { - const batch = this.takePending(); - if (batch.length === 0) { - return; - } - for (const entry of batch) { - this.inFlight.set(entryKey(entry), entry); - } - this.writesInFlight += 1; - try { - await Promise.all(batch.map((entry) => this.send(entry))); - } finally { - this.writesInFlight -= 1; - } - }); - this.sending = run.then( - () => undefined, - () => undefined - ); - return this.sending; - } - - /** - * Page hide: `keepalive` requests survive the page. They go out at once - * unless a write is still travelling, which they must not overtake, and - * later writes wait for them in turn. - */ - flushWithKeepalive(): void { - const entries = this.takePending(); - if (entries.length === 0) { - // Writes still travelling may be cut off with the page. - this.saveUnsent(); - return; - } - const sendAll = async (): Promise => { - this.writesInFlight += 1; - try { - await Promise.all( - entries.map((entry) => this.sendKeepalive(entry)) - ); - } finally { - this.writesInFlight -= 1; - } - }; - for (const entry of entries) { - this.inFlight.set(entryKey(entry), entry); - } - this.saveUnsent(); - // Idle: send synchronously, while the hiding page still runs. - const run = - this.writesInFlight === 0 ? sendAll() : this.sending.then(sendAll); - this.sending = run.then( - () => undefined, - () => undefined - ); - } - private readCache(namespace: ServerUserStateNamespace): string | null { return ( this.overflow.get(namespace) ?? @@ -297,21 +253,34 @@ export class ServerUserDataStorage implements UserDataStorage { } } + /** A conflict replay settled what the entry holds. */ + private showEntry( + namespace: ServerUserStateNamespace, + playlistId: string, + value: unknown + ): void { + const map = this.baseline(namespace); + if (value === undefined) { + delete map[playlistId]; + } else { + map[playlistId] = value; + } + this.writeCache(namespace, JSON.stringify(map)); + } + private isSynced(key: string): key is ServerUserStateNamespace { 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. + * What a write is compared against: the map the caller read, so an edit + * sends only its own entries, never another tab's changes in the shared + * cache. */ private baseline( namespace: ServerUserStateNamespace ): Record { - return ( - this.synced.get(namespace) ?? - parseMap(this.readCache(namespace) ?? '{}') - ); + return parseMap(this.readCache(namespace) ?? '{}'); } private queueChanges( @@ -321,7 +290,12 @@ export class ServerUserDataStorage implements UserDataStorage { ): void { const change = (playlistId: string, value: unknown) => { if (this.hydrated) { - this.queue(namespace, playlistId, value); + this.writes.queue( + namespace, + playlistId, + value, + previous[playlistId] + ); return; } const key = `${namespace}|${playlistId}`; @@ -350,107 +324,13 @@ export class ServerUserDataStorage implements UserDataStorage { change(playlistId, undefined); } } - this.synced.set(namespace, next); - this.scheduleFlush(); - } - - 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 { - if (this.pending.size === 0) { - return; - } - // Saved at once: another tab's re-read must not replace them. - this.saveUnsent(); - if (this.flushTimer !== null) { - return; - } - this.flushTimer = setTimeout(() => { - this.flushTimer = null; - void this.flush(); - }, FLUSH_DELAY_MS); - } - - private takePending(): PendingEntry[] { - if (this.flushTimer !== null) { - clearTimeout(this.flushTimer); - this.flushTimer = null; - } - const batch = [...this.pending.values()]; - this.pending.clear(); - 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): void { - const key = entryKey(entry); - if (this.edits.get(key) === entry.edit && !this.pending.has(key)) { - this.pending.set(key, entry); - } - } - - private async sendKeepalive(entry: PendingEntry): Promise { - const path = entryPath(entry); - const ok = await (entry.value === undefined - ? this.api.sendKeepalive('DELETE', path) - : this.api.sendKeepalive('PUT', path, { value: entry.value })); - this.retireInFlight(entry); - if (!ok) { - this.requeue(entry); - } - this.saveUnsent(); - } - - private async send(entry: PendingEntry): Promise { - try { - if (entry.value === undefined) { - await this.api.request(entryPath(entry), { method: 'DELETE' }); - } else { - await this.api.sendJson('PUT', entryPath(entry), { - value: entry.value, - }); - } - this.retireInFlight(entry); - this.saveUnsent(); - } catch (error) { - this.retireInFlight(entry); - this.requeue(entry); - this.saveUnsent(); - this.onWriteFailure(error); - } - } - - /** A newer write of the same entry may have replaced it meanwhile. */ - private retireInFlight(entry: PendingEntry): void { - const key = entryKey(entry); - if (this.inFlight.get(key)?.edit === entry.edit) { - this.inFlight.delete(key); - } + this.writes.scheduleFlush(); } /** Persists what the server has not confirmed; skips idle rewrites. */ private saveUnsent(): void { // A newer edit can be queued while an older one is still travelling. - const writes = newestWrites([ - ...this.inFlight.values(), - ...this.pending.values(), - ]); + const writes = this.writes.unsentWrites(); if (writes.length === 0 && !this.unsentSaved) { return; } @@ -461,8 +341,7 @@ export class ServerUserDataStorage implements UserDataStorage { private publishUnconfirmed(): void { markUnconfirmed(this.local, this.instanceId, this.tabId, [ - ...this.pending.keys(), - ...this.inFlight.keys(), + ...this.writes.unconfirmedKeys(), ...this.deferred.keys(), ]); } diff --git a/libs/services/src/lib/server-storage/server-user-data-sync.spec.ts b/libs/services/src/lib/server-storage/server-user-data-sync.spec.ts new file mode 100644 index 000000000..c1c4cbc22 --- /dev/null +++ b/libs/services/src/lib/server-storage/server-user-data-sync.spec.ts @@ -0,0 +1,231 @@ +import { ServerApiError } from './server-api-client'; +import { ServerUserDataStorage } from './server-user-data-storage'; + +function memoryStorage() { + const values = new Map(); + return { + values, + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => void values.set(key, value), + removeItem: (key: string) => void values.delete(key), + }; +} + +/** A server that keeps entries, tags them by content and refuses stale tags. */ +function fakeServer(initial: Record> = {}) { + const state: Record> = { + 'xtream-favorites': {}, + 'xtream-recent-items': {}, + 'xtream-collection-items': {}, + 'xtream-playback-positions': {}, + ...structuredClone(initial), + }; + const tagOf = (value: unknown) => + value === undefined + ? null + : `t${JSON.stringify(value).replace(/"/g, "'")}`; + const answer = (body: unknown) => + ({ json: async () => body }) as unknown as Response; + const write = ( + path: string, + headers: Record, + value?: unknown + ) => { + const [, , namespace, playlistId] = path.split('/'); + const current = state[namespace][playlistId]; + const expected = + headers['If-None-Match'] === '*' + ? null + : headers['If-Match']?.replace(/^"|"$/g, ''); + if (expected !== undefined && expected !== tagOf(current)) { + throw new ServerApiError(412, 'conflict', 'changed', { + ...(current === undefined ? {} : { value: current }), + tag: tagOf(current), + }); + } + if (value === undefined) { + delete state[namespace][playlistId]; + } else { + state[namespace][playlistId] = value; + } + return answer({ tag: tagOf(value) }); + }; + const api = { + getJson: jest.fn(async () => { + const tags: Record> = {}; + for (const [namespace, entries] of Object.entries(state)) { + tags[namespace] = Object.fromEntries( + Object.entries(entries).map(([id, value]) => [ + id, + tagOf(value) as string, + ]) + ); + } + return { entries: structuredClone(state), tags }; + }), + sendJson: jest.fn( + async ( + _method: string, + path: string, + body: { value: unknown }, + headers: Record = {} + ) => write(path, headers, body.value) + ), + request: jest.fn( + async (path: string, init: { headers?: Record }) => + write(path, init.headers ?? {}) + ), + readJson: jest.fn((response: Response) => response.json()), + sendKeepalive: jest.fn().mockResolvedValue(true), + }; + return { state, api }; +} + +const favorites = (storage: ServerUserDataStorage) => + JSON.parse(storage.getItem('xtream-favorites') ?? '{}'); + +describe('ServerUserDataStorage with other devices', () => { + it('replays an edit on what another device saved meanwhile', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['live:1'] } }); + const phone = new ServerUserDataStorage( + server.api as never, + memoryStorage(), + jest.fn() + ); + await phone.hydrate(); + + // The TV adds live:2 after the phone read the entry. + server.state['xtream-favorites']['p1'] = ['live:1', 'live:2']; + phone.setItem( + 'xtream-favorites', + JSON.stringify({ p1: ['live:1', 'live:9'] }) + ); + await phone.flush(); + await phone.flush(); + + expect(server.state['xtream-favorites']['p1']).toEqual([ + 'live:1', + 'live:2', + 'live:9', + ]); + expect(favorites(phone)).toEqual({ + p1: ['live:1', 'live:2', 'live:9'], + }); + }); + + it('replays a newer queued edit together with the refused one', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const phone = new ServerUserDataStorage( + server.api as never, + memoryStorage(), + jest.fn() + ); + await phone.hydrate(); + server.state['xtream-favorites']['p1'] = ['S', 'Z']; + + phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + const first = phone.flush(); + phone.setItem( + 'xtream-favorites', + JSON.stringify({ p1: ['S', 'X', 'Y'] }) + ); + await first; + await phone.flush(); + await phone.flush(); + + expect(server.state['xtream-favorites']['p1']).toEqual([ + 'S', + 'Z', + 'X', + 'Y', + ]); + }); + + it('builds a queued edit on the version its predecessor created', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const phone = new ServerUserDataStorage( + server.api as never, + memoryStorage(), + jest.fn() + ); + await phone.hydrate(); + + let answer: () => void = () => undefined; + const send = server.api.sendJson.getMockImplementation(); + if (!send) { + throw new Error('The fake server answers sendJson'); + } + server.api.sendJson.mockImplementationOnce( + (...args: Parameters) => + new Promise( + (resolve) => (answer = () => resolve(send(...args))) + ) + ); + phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + const first = phone.flush(); + await new Promise((resolve) => setTimeout(resolve, 0)); + // The first write is travelling when the second edit comes in. + phone.setItem( + 'xtream-favorites', + JSON.stringify({ p1: ['S', 'X', 'Y'] }) + ); + answer(); + await first; + await phone.flush(); + + expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'X', 'Y']); + // No conflict round trip: the second write carried the new tag. + expect(server.api.sendJson).toHaveBeenCalledTimes(2); + }); + + it("never sends another tab's change to an entry it did not edit", async () => { + const server = fakeServer({ + 'xtream-favorites': { p1: ['a'], p2: ['b'] }, + }); + const shared = memoryStorage(); + const tab = () => + new ServerUserDataStorage( + server.api as never, + shared, + jest.fn(), + 'default', + memoryStorage() + ); + const tabA = tab(); + const tabB = tab(); + await tabA.hydrate(); + await tabB.hydrate(); + + tabB.setItem( + 'xtream-favorites', + JSON.stringify({ ...favorites(tabB), p1: ['a', 'b1'] }) + ); + tabA.setItem( + 'xtream-favorites', + JSON.stringify({ ...favorites(tabA), p2: ['b', 'a2'] }) + ); + await tabA.flush(); + + expect(server.api.sendJson.mock.calls.map(([, path]) => path)).toEqual([ + '/user-state/xtream-favorites/p2', + ]); + }); + + it('drops queued edits and cache before a playlist is deleted', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['a'] } }); + const phone = new ServerUserDataStorage( + server.api as never, + memoryStorage(), + jest.fn() + ); + await phone.hydrate(); + phone.setItem('xtream-favorites', JSON.stringify({ p1: ['a', 'b'] })); + + await phone.discard('p1'); + await phone.flush(); + + expect(server.api.sendJson).not.toHaveBeenCalled(); + expect(favorites(phone)).toEqual({}); + expect(phone.hasUnsentWrites()).toBe(false); + }); +}); diff --git a/libs/shared/interfaces/src/lib/server-storage.contract.ts b/libs/shared/interfaces/src/lib/server-storage.contract.ts index a9364be61..f1e7359fb 100644 --- a/libs/shared/interfaces/src/lib/server-storage.contract.ts +++ b/libs/shared/interfaces/src/lib/server-storage.contract.ts @@ -58,10 +58,30 @@ export interface ServerStorageStatusResponse { export interface ServerUserStateResponse { readonly entries: Record>; + /** + * Version tag of every entry. A write sends the tag it is based on + * (`If-Match: ""`, or `If-None-Match: *` for an entry the client + * saw absent); a stale tag gets `412` with the current entry. + */ + readonly tags?: Record>; +} + +/** Answer to a user-state write: the entry's new tag, `null` once deleted. */ +export interface ServerUserStateWriteResponse { + readonly tag: string | null; +} + +/** The entry as the server has it, sent with a `412` conflict. */ +export interface ServerUserStateCurrent { + /** Absent when the entry does not exist. */ + readonly value?: unknown; + readonly tag: string | null; } export interface ServerStorageErrorBody { readonly message: string; readonly status: number; readonly code?: string; + /** With `code: 'conflict'` (412): the user-state entry as it is now. */ + readonly current?: ServerUserStateCurrent; }