fix(server-storage): conditional user-data writes, scoped token, deletes

- User-state entries carry a content tag; PUT/DELETE accept If-Match or
  If-None-Match: * and answer 412 with the current value and tag. The PWA
  sends each edit on the tag it read and replays it (before → after) on
  the server's value after a conflict, so another device's change to the
  same playlist's favorites or positions is kept instead of overwritten.
- The access token goes only to the backend URL (same origin and path),
  never to another host or a sibling path.
- Deleting a playlist (or all) first drops its queued user-data edits and
  waits for writes in flight, so none recreates deleted data.
- The migration copy writes on the tags it read.
- The write queue moves out of ServerUserDataStorage into
  UserDataWriteQueue.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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<ServerUserStateResponse> {
+        const tags = Object.fromEntries(
+            SERVER_USER_STATE_NAMESPACES.map((namespace) => [
+                namespace,
+                Object.create(null) as Record<string, string>,
+            ])
+        ) as Required<ServerUserStateResponse>['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 {

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) => {

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: "<tag>"` 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(

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
+        );
+    });
+});

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<unknown>,
         next: HttpHandler
     ): Observable<HttpEvent<unknown>> {
-        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;
+    }
+}

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<string, string> = {}
     ): Promise<Response> {
         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<string, string> = {}
     ): Promise<boolean> {
         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<Pick<ServerStorageErrorBody, 'message' | 'code' | 'current'>> {
     try {
         const body = (await response.json()) as Partial<ServerStorageErrorBody>;
         return {
             message: body.message ?? response.statusText,
             code: body.code,
+            current: body.current,
         };
     } catch {
         return { message: response.statusText };

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'] },
+                {},
             ],
         ]);
     });

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<ServerUserStateResponse>(
                 '/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;

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',
+        ]);
+    });
 });

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<object, string>();
 
-    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<void> = async () => undefined
+    ) {}
 
     dbGetAppPlaylists(): Promise<Playlist[]> {
         return this.api.getJson<Playlist[]>('/playlists');
@@ -111,6 +120,7 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore {
     }
 
     async dbDeletePlaylist(playlistId: string): Promise<ElectronBridgeResult> {
+        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<ElectronBridgeResult> {
+        await this.beforeDelete();
         await this.api.request('/playlists', { method: 'DELETE' });
         this.cache.clear();
         return { success: true };

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.

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 {

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<string, string> {
+    if (tag === undefined) {
+        return {};
+    }
+    return tag === null ? { 'If-None-Match': '*' } : { 'If-Match': `"${tag}"` };
 }
 
 export function entryKey(entry: PendingEntry): string {

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<string, string>
+    >();
+    private readonly pending = new Map<string, PendingEntry>();
+    /** Writes still travelling (normal or page-hide), by entry key. */
+    private readonly inFlight = new Map<string, PendingEntry>();
+    private flushTimer: ReturnType<typeof setTimeout> | null = null;
+    /** Batches go out one at a time, so an older write never lands last. */
+    private sending: Promise<void> = Promise.resolve();
+    private writesInFlight = 0;
+    /** Edit counter per `namespace|playlistId`, for re-reads to keep them. */
+    private readonly edits = new Map<string, number>();
+    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<string> {
+        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<Record<string, string>>
+    ): 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<void> {
+        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<void> => {
+            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<void> {
+        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<void> {
+        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<void> {
+        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<Partial<ServerUserStateWriteResponse>>(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();
+    }
+}

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'] },
+            {}
         );
     });
 });

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<string, unknown>
-    >();
-    private readonly pending = new Map<string, PendingEntry>();
-    private flushTimer: ReturnType<typeof setTimeout> | null = null;
-    /** Batches go out one at a time, so an older write never lands last. */
-    private sending: Promise<void> = Promise.resolve();
-    private writesInFlight = 0;
-    /** Writes still travelling (normal or page-hide), by entry key. */
-    private readonly inFlight = new Map<string, PendingEntry>();
+    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<ServerUserStateNamespace, string>();
     /** Re-reads run one at a time, so an older one never lands last. */
     private hydration: Promise<void> = Promise.resolve();
-    /** Edit counter per `namespace|playlistId`, for hydrate to keep them. */
-    private readonly edits = new Map<string, number>();
-    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<void> {
-        // 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<ServerUserStateResponse>('/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<void> {
+        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<void> {
+        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<void> {
+        // 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<ServerUserStateResponse>('/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<void> {
-        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<void> => {
-            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<string, unknown> {
-        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<void> {
-        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<void> {
-        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(),
         ]);
     }

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<string, string>();
+    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<string, Record<string, unknown>> = {}) {
+    const state: Record<string, Record<string, unknown>> = {
+        '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<string, string>,
+        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<string, Record<string, string>> = {};
+            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<string, string> = {}
+            ) => write(path, headers, body.value)
+        ),
+        request: jest.fn(
+            async (path: string, init: { headers?: Record<string, string> }) =>
+                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<typeof send>) =>
+                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);
+    });
+});

libs/shared/interfaces/src/lib/server-storage.contract.ts

@@ -58,10 +58,30 @@ export interface ServerStorageStatusResponse {
 
 export interface ServerUserStateResponse {
     readonly entries: Record<ServerUserStateNamespace, Record<string, unknown>>;
+    /**
+     * Version tag of every entry. A write sends the tag it is based on
+     * (`If-Match: "<tag>"`, or `If-None-Match: *` for an entry the client
+     * saw absent); a stale tag gets `412` with the current entry.
+     */
+    readonly tags?: Record<ServerUserStateNamespace, Record<string, string>>;
+}
+
+/** 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;
 }