fix(server-storage): keep steps of possibly-landed writes apart, queue closed tabs' edits even when the cache shows them

- Steps of a write whose answer was lost, of saved writes taken over after
  a reload, and of a closed tab's saved edits are marked uncertain, so a
  later edit never joins them: adding and then removing an item after a
  lost answer still removes it.
- A closed tab's saved edits are queued ahead of this tab's own even when
  the cache already shows them, so a conflict replays them too.
- The tag map moves into UserStateTags.

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

docs/architecture/pwa-self-hosted.md

@@ -210,8 +210,8 @@ settings and guards read them in every build.
   per page load (`iptvnator-server:unsent-writes:<instanceId>:<page>`). The
   tab remembers its page load in `sessionStorage`, so a reload takes its
   writes over and sends them first; the writes of a tab closed before it
-  could send them are replayed by the next re-read in another tab once
-  that tab's record has gone stale. Tabs never clear each other's.
+  could send them are queued, ahead of its own, by the next re-read in
+  another tab once that tab's record has gone stale. Tabs never clear each other's.
   Cache and saved writes belong to one server database
   (`iptvnator-server:cache-owner`); another one behind the same address
   starts clean. Also,
@@ -222,8 +222,9 @@ settings and guards read them in every build.
   replayed on the server's current value, so another device's change to the
   same playlist's entry is kept. A queued write keeps its edits as steps
   (before → after, also in the saved writes): an edit continuing the last
-  one joins it, one made behind a travelling write or on another tab's
-  change is kept apart, and a conflict replays them in order. So a lost
+  one joins it; one made behind a travelling write, after a write whose
+  answer was lost, or on another tab's change is kept apart; and a
+  conflict replays them in order. So a lost
   write is not dropped, and adding and then removing an item still removes
   it when the addition landed. When two tabs edit one entry before either
   write lands, a removal of the other tab's unsent item holds if that tab's

libs/services/src/lib/server-storage/server-user-data-entries.ts

@@ -27,10 +27,19 @@ export interface PendingEntry {
 export interface EditStep {
     readonly before: unknown;
     readonly after: unknown;
-    /** The travelling write that carries it, until that one is answered. */
+    /**
+     * The travelling write that carries it, until that one is answered;
+     * `0` once a write that carried it was lost or saved for a later page:
+     * it may have landed, so a later edit never joins it.
+     */
     readonly sent?: number;
 }
 
+/** Steps whose write may or may not have landed. */
+export function uncertain(steps: readonly EditStep[]): EditStep[] {
+    return steps.map((step) => ({ ...step, sent: 0 }));
+}
+
 /**
  * `steps` plus the edit `before` → `after`: joined with the last step when
  * it continues it, kept apart otherwise.

libs/services/src/lib/server-storage/server-user-data-queue.ts

@@ -11,9 +11,11 @@ import {
     type EditStep,
     type PendingEntry,
     preconditionHeaders,
+    uncertain,
     UNKNOWN_TAG,
 } from './server-user-data-entries';
 import { replaySteps } from './server-user-data-edit';
+import { UserStateTags } from './server-user-data-tags';
 import { newestWrites } from './server-unsent-writes';
 
 const FLUSH_DELAY_MS = 500;
@@ -44,10 +46,7 @@ export interface UserDataQueueHooks {
  */
 export class UserDataWriteQueue {
     /** Server version tag per entry; absent until the first read. */
-    private readonly tags = new Map<
-        ServerUserStateNamespace,
-        Record<string, string>
-    >();
+    private readonly tags = new UserStateTags();
     private readonly pending = new Map<string, PendingEntry>();
     /** Writes still travelling (normal or page-hide), by entry key. */
     private readonly inFlight = new Map<string, PendingEntry>();
@@ -114,7 +113,7 @@ export class UserDataWriteQueue {
         namespace: ServerUserStateNamespace,
         tags: Readonly<Record<string, string>>
     ): void {
-        this.tags.set(namespace, { ...tags });
+        this.tags.set(namespace, tags);
     }
 
     /**
@@ -168,8 +167,10 @@ export class UserDataWriteQueue {
                 tag:
                     base.tag !== undefined
                         ? base.tag
-                        : this.tagFor(namespace, playlistId),
-                steps: base.steps ?? [{ before, after: value }],
+                        : this.tags.of(namespace, playlistId),
+                steps: base.steps
+                    ? uncertain(base.steps)
+                    : [{ before, after: value }],
             };
         }
         if (base.foreign) {
@@ -179,6 +180,28 @@ export class UserDataWriteQueue {
         this.edits.set(key, edit);
     }
 
+    /**
+     * A closed tab's saved edits, ahead of this tab's own (which may have
+     * been made on them): sent with them, replayed first on a conflict.
+     */
+    adopt(
+        namespace: ServerUserStateNamespace,
+        playlistId: string,
+        value: unknown,
+        steps: readonly EditStep[],
+        foreign: boolean
+    ): void {
+        this.queue(namespace, playlistId, value, value, { foreign });
+        const key = `${namespace}|${playlistId}`;
+        const entry = this.pending.get(key);
+        if (entry) {
+            this.pending.set(key, {
+                ...entry,
+                steps: [...uncertain(steps), ...entry.steps],
+            });
+        }
+    }
+
     scheduleFlush(): void {
         if (this.pending.size === 0) {
             return;
@@ -271,41 +294,10 @@ export class UserDataWriteQueue {
                 }
             }
         }
-        for (const [namespace, tags] of this.tags) {
-            if (playlistId === undefined) {
-                this.tags.set(namespace, {});
-            } else {
-                delete tags[playlistId];
-            }
-        }
+        this.tags.forget(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);
@@ -328,7 +320,7 @@ export class UserDataWriteQueue {
     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);
+            this.pending.set(key, { ...entry, steps: uncertain(entry.steps) });
         }
     }
 
@@ -424,7 +416,7 @@ export class UserDataWriteQueue {
 
     /** The server took `entry`: the edit queued behind it builds on it. */
     private confirmed(entry: PendingEntry, tag: string | null): void {
-        this.setTag(entry.namespace, entry.playlistId, tag);
+        this.tags.update(entry.namespace, entry.playlistId, tag);
         const key = entryKey(entry);
         const next = this.pending.get(key);
         if (next?.steps.some((step) => step.sent === entry.edit)) {
@@ -446,7 +438,7 @@ export class UserDataWriteQueue {
         entry: PendingEntry,
         current: ServerUserStateCurrent
     ): void {
-        this.setTag(entry.namespace, entry.playlistId, current.tag);
+        this.tags.update(entry.namespace, entry.playlistId, current.tag);
         const key = entryKey(entry);
         const latest =
             this.pending.get(key) ??

libs/services/src/lib/server-storage/server-user-data-storage.ts

@@ -122,8 +122,8 @@ export class ServerUserDataStorage implements UserDataStorage {
     }
 
     /**
-     * Saved writes and early edits of closed tabs: replayed as this tab's
-     * own edits on what it shows now, then forgotten there.
+     * Saved writes and early edits of closed tabs: shown on what this tab
+     * shows now and queued with its writes, then forgotten there.
      */
     private adoptClosedTabs(): void {
         for (const page of orphanedPages(
@@ -135,18 +135,28 @@ export class ServerUserDataStorage implements UserDataStorage {
                 ...readDeferredEdits(this.local, this.instanceId, page),
                 ...readUnsentWrites(this.local, this.instanceId, page),
             ];
-            for (const edit of edits) {
-                const before = this.baseline(edit.namespace);
-                const after = { ...before };
-                const value = replaySteps(before[edit.playlistId], edit.steps);
-                if (value === undefined) {
-                    delete after[edit.playlistId];
-                } else {
-                    after[edit.playlistId] = value;
-                }
-                this.writeCache(edit.namespace, JSON.stringify(after));
-                this.queueChanges(edit.namespace, after, before);
+            const foreign = othersUnconfirmed(
+                this.local,
+                this.instanceId,
+                this.tabId
+            );
+            for (const { namespace, playlistId, steps } of edits) {
+                const value = replaySteps(
+                    this.baseline(namespace)[playlistId],
+                    steps
+                );
+                this.showEntry(namespace, playlistId, value);
+                // Queued even when the cache shows it already.
+                this.writes.adopt(
+                    namespace,
+                    playlistId,
+                    value,
+                    steps,
+                    foreign.has(`${namespace}|${playlistId}`)
+                );
             }
+            // Saved here before the closed tab's copy is forgotten.
+            this.writes.scheduleFlush();
             removePage(this.local, this.instanceId, page);
         }
     }

libs/services/src/lib/server-storage/server-user-data-sync.spec.ts

@@ -1,4 +1,5 @@
 import { ServerApiError } from './server-api-client';
+import { UNCONFIRMED_TTL_MS } from './server-unsent-writes';
 import { ServerUserDataStorage } from './server-user-data-storage';
 
 function memoryStorage() {
@@ -593,4 +594,88 @@ describe('ServerUserDataStorage with other devices', () => {
 
         expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'X']);
     });
+
+    it('keeps a removal made after a write whose answer was lost', async () => {
+        const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } });
+        const phone = new ServerUserDataStorage(
+            server.api as never,
+            memoryStorage(),
+            jest.fn()
+        );
+        await phone.hydrate();
+        const send = server.api.sendJson.getMockImplementation();
+        if (!send) {
+            throw new Error('The fake server answers sendJson');
+        }
+        // The server saves X, but the answer never arrives.
+        server.api.sendJson.mockImplementationOnce(
+            async (...args: Parameters<typeof send>) => {
+                await send(...args);
+                throw new ServerApiError(0, 'network', 'offline');
+            }
+        );
+        phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] }));
+        await phone.flush();
+        expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'X']);
+
+        phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S'] }));
+        await phone.flush();
+        await phone.flush();
+
+        expect(server.state['xtream-favorites']['p1']).toEqual(['S']);
+    });
+
+    it("sends a closed tab's favorite this tab's cache already shows", async () => {
+        const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } });
+        const shared = memoryStorage();
+        const send = server.api.sendJson.getMockImplementation();
+        if (!send) {
+            throw new Error('The fake server answers sendJson');
+        }
+        const offline = () =>
+            Promise.reject(new ServerApiError(0, 'network', 'offline'));
+        let online = false;
+        const tabApi = {
+            ...server.api,
+            sendJson: jest.fn((...args: Parameters<typeof send>) =>
+                online ? send(...args) : offline()
+            ),
+        };
+        const clock = jest.spyOn(Date, 'now').mockReturnValue(1_000);
+        const closed = new ServerUserDataStorage(
+            { ...server.api, sendJson: jest.fn(offline) } as never,
+            shared,
+            jest.fn(),
+            'default',
+            memoryStorage()
+        );
+        await closed.hydrate();
+        closed.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] }));
+        await closed.flush();
+        const tab = new ServerUserDataStorage(
+            tabApi as never,
+            shared,
+            jest.fn(),
+            'default',
+            memoryStorage()
+        );
+        await tab.hydrate();
+        tab.setItem(
+            'xtream-favorites',
+            JSON.stringify({ p1: [...favorites(tab).p1, 'Y'] })
+        );
+        await tab.flush();
+
+        // Later the closed tab's record is stale; this tab is still offline.
+        clock.mockReturnValue(1_000 + UNCONFIRMED_TTL_MS + 1);
+        await tab.hydrate();
+        online = true;
+        await tab.flush();
+        await tab.flush();
+        clock.mockRestore();
+
+        expect(
+            [...(server.state['xtream-favorites']['p1'] as string[])].sort()
+        ).toEqual(['S', 'X', 'Y']);
+    });
 });

libs/services/src/lib/server-storage/server-user-data-tags.ts

@@ -0,0 +1,54 @@
+import type { ServerUserStateNamespace } from '@iptvnator/shared/interfaces';
+
+/** The server version tag of each user-state entry, by namespace. */
+export class UserStateTags {
+    /** Absent for a namespace until the first read. */
+    private readonly tags = new Map<
+        ServerUserStateNamespace,
+        Record<string, string>
+    >();
+
+    set(
+        namespace: ServerUserStateNamespace,
+        tags: Readonly<Record<string, string>>
+    ): void {
+        this.tags.set(namespace, { ...tags });
+    }
+
+    /** `undefined` until the server's tags are known (or not offered). */
+    of(
+        namespace: ServerUserStateNamespace,
+        playlistId: string
+    ): string | null | undefined {
+        const tags = this.tags.get(namespace);
+        return tags ? (tags[playlistId] ?? null) : undefined;
+    }
+
+    /** `null`: the entry is gone. */
+    update(
+        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;
+        }
+    }
+
+    /** After a delete: one playlist's tags, without an id all of them. */
+    forget(playlistId?: string): void {
+        for (const [namespace, tags] of this.tags) {
+            if (playlistId === undefined) {
+                this.tags.set(namespace, {});
+            } else {
+                delete tags[playlistId];
+            }
+        }
+    }
+}