fix(server-storage): favorite toggles as deltas, early edits as steps on recovered writes, release kept journal copies

- PlaylistEffects.updateFavorites$ applies the toggled channel through
  transformPlaylistFavorites, on the favorites read at write time, instead
  of writing the reducer's whole array: a conflict retry keeps another
  device's favorites (#1919).
- Edits made before the first read on top of this tab's recovered unsent
  write are appended to it as their own steps, so a removal of an item the
  recovered write adds survives a conflict replay.
- Journal copies of earlier page loads kept because this page could not
  save its own are removed once it can.
- adopt/extend move into server-user-data-adopt.ts over a small queue API
  (unsent, isQueued, put); the queue exposes its tag map.

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

docs/architecture/pwa-self-hosted.md

@@ -204,7 +204,9 @@ settings and guards read them in every build.
   throws `PlaylistWriteConflict`, and `PlaylistsService`'s per-playlist
   write queue runs the mutation again on a fresh read (a retry on the same,
   unread copy writes unconditionally, as before). Imports and start-up
-  migrations (`dbUpsertAppPlaylists`) write unconditionally. After a write the store
+  migrations (`dbUpsertAppPlaylists`) write unconditionally. An M3U favorite
+  toggle goes through `transformPlaylistFavorites`, adding or removing that
+  one channel on the favorites read at write time. After a write the store
   keeps its copy only when `X-Replaced-ETag` names the version its payload
   came from.
 - **Xtream user data.** `USER_DATA_STORAGE` becomes `ServerUserDataStorage`:
@@ -220,7 +222,7 @@ 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 old copy goes only once the new one
-  is saved); the writes of a tab closed before it
+  is saved, at the latest with this page's next saved copy); the writes of a tab closed before it
   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; the result replays the
   closed tab's edits first and this tab's newer ones last. Tabs never clear each other's.
@@ -249,9 +251,9 @@ settings and guards read them in every build.
   being deleted its edits wait; the delete drops them only when it
   succeeds. Edits made before the first successful read
   are held as before → after and applied to the server's value once it
-  arrives (`applyEdit`), or to this tab's recovered unsent write when it has
-  one: a new device or a stale cache adds its favorite instead of replacing
-  the server's list. They are saved per page load too
+  arrives (`applyEdit`): a new device or a stale cache adds its favorite
+  instead of replacing the server's list. On this tab's recovered unsent
+  write they are kept as their own steps after it. They are saved per page load too
   (`iptvnator-server:deferred-edits:<instanceId>:<page>`). A shared
   registry in `localStorage` (`iptvnator-server:unconfirmed:<instanceId>`,
   one record per page load, renewed every 30 s while it has unsent writes,

libs/m3u-state/src/lib/effects.ts

@@ -89,9 +89,21 @@ export class PlaylistEffects {
                     this.store.select(selectActivePlaylistId)
                 ),
                 filter(([, , playlistId]) => !!playlistId),
-                switchMap(([, favorites, playlistId]) =>
-                    this.playlistsService.updateFavorites(playlistId, favorites)
-                ),
+                // The toggle, applied to the favorites as stored at write
+                // time: another device's changes to other channels stay, also
+                // when a conflict makes the write run again on a fresh read.
+                mergeMap(([{ channel }, favorites, playlistId]) => {
+                    const added = favorites.includes(channel.url);
+                    return this.playlistsService.transformPlaylistFavorites(
+                        playlistId,
+                        (current) => {
+                            const others = current.filter(
+                                (favorite) => favorite !== channel.url
+                            );
+                            return added ? [...others, channel.url] : others;
+                        }
+                    );
+                }),
                 tap(() => {
                     this.snackBar.open(
                         this.translate.instant('CHANNELS.FAVORITES_UPDATED'),

libs/m3u-state/src/lib/favorites.effect.spec.ts

@@ -0,0 +1,93 @@
+import { Injector, runInInjectionContext } from '@angular/core';
+import { MatSnackBar } from '@angular/material/snack-bar';
+import { Router } from '@angular/router';
+import { Actions } from '@ngrx/effects';
+import { Store } from '@ngrx/store';
+import { TranslateService } from '@ngx-translate/core';
+import { EpgService } from '@iptvnator/epg/data-access';
+import { PlaylistsService, SettingsStore } from '@iptvnator/services';
+import type { Channel } from '@iptvnator/shared/interfaces';
+import { of, Subject } from 'rxjs';
+import { FavoritesActions } from './actions';
+import { PlaylistEffects } from './effects';
+import { selectActivePlaylistId, selectFavorites } from './selectors';
+
+describe('PlaylistEffects updateFavorites$', () => {
+    let actions$: Subject<unknown>;
+    let transformPlaylistFavorites: jest.Mock;
+    let storedFavorites: string[];
+
+    function setup(afterToggle: string[]) {
+        actions$ = new Subject();
+        transformPlaylistFavorites = jest.fn(
+            (_id: string, transform: (current: string[]) => string[]) => {
+                storedFavorites = transform(storedFavorites);
+                return of(undefined);
+            }
+        );
+        const injector = Injector.create({
+            parent: { get: () => ({}) } as unknown as Injector,
+            providers: [
+                { provide: Actions, useValue: new Actions(actions$) },
+                {
+                    provide: Store,
+                    useValue: {
+                        select: (selector: unknown) =>
+                            of(
+                                selector === selectFavorites
+                                    ? afterToggle
+                                    : selector === selectActivePlaylistId
+                                      ? 'p1'
+                                      : undefined
+                            ),
+                    },
+                },
+                {
+                    provide: PlaylistsService,
+                    useValue: { transformPlaylistFavorites },
+                },
+                { provide: EpgService, useValue: { fetchEpg: jest.fn() } },
+                { provide: Router, useValue: {} },
+                { provide: MatSnackBar, useValue: { open: jest.fn() } },
+                {
+                    provide: TranslateService,
+                    useValue: { instant: (key: string) => key },
+                },
+                {
+                    provide: SettingsStore,
+                    useValue: { getSettings: () => ({ epgUrl: [] }) },
+                },
+            ],
+        });
+        const effects = runInInjectionContext(
+            injector,
+            () => new PlaylistEffects()
+        );
+        effects.updateFavorites$.subscribe();
+    }
+
+    const channel = { url: 'http://a' } as Channel;
+
+    it("adds the toggled channel to the stored favorites, keeping others'", () => {
+        // This device's state did not know about 'http://b' yet.
+        storedFavorites = ['http://b'];
+        setup(['http://a']);
+
+        actions$.next(FavoritesActions.updateFavorites({ channel }));
+
+        expect(transformPlaylistFavorites).toHaveBeenCalledWith(
+            'p1',
+            expect.any(Function)
+        );
+        expect(storedFavorites).toEqual(['http://b', 'http://a']);
+    });
+
+    it('removes only the toggled channel', () => {
+        storedFavorites = ['http://a', 'http://b'];
+        setup([]);
+
+        actions$.next(FavoritesActions.updateFavorites({ channel }));
+
+        expect(storedFavorites).toEqual(['http://b']);
+    });
+});

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

@@ -0,0 +1,68 @@
+import type { ServerUserStateNamespace } from '@iptvnator/shared/interfaces';
+import {
+    behind,
+    type EditStep,
+    uncertain,
+    UNKNOWN_TAG,
+} from './server-user-data-entries';
+import { replaySteps } from './server-user-data-edit';
+import type { UserDataWriteQueue } from './server-user-data-queue';
+
+/**
+ * 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. Returns the
+ * value to show.
+ */
+export function adoptEdits(
+    writes: UserDataWriteQueue,
+    namespace: ServerUserStateNamespace,
+    playlistId: string,
+    current: unknown,
+    steps: readonly EditStep[],
+    foreign: boolean
+): unknown {
+    const key = `${namespace}|${playlistId}`;
+    const adopted = uncertain(steps);
+    const queued = writes.isQueued(key) ? writes.unsent(key) : undefined;
+    if (queued) {
+        // This tab's queued edits are newer: they replay last.
+        const all = [...adopted, ...queued.steps];
+        const value = replaySteps(queued.steps[0]?.before, all);
+        writes.put({
+            ...queued,
+            value,
+            steps: all,
+            tag: foreign ? UNKNOWN_TAG : queued.tag,
+        });
+        return value;
+    }
+    const value = replaySteps(current, steps);
+    writes.queue(namespace, playlistId, value, value, { foreign });
+    const entry = writes.unsent(key);
+    if (entry) {
+        writes.put({ ...entry, steps: [...adopted, ...entry.steps] });
+    }
+    return value;
+}
+
+/**
+ * Early edits made on this tab's own unsent write: kept as their own steps
+ * after it, so a conflict replays them in order.
+ */
+export function extendUnsent(
+    writes: UserDataWriteQueue,
+    key: string,
+    steps: readonly EditStep[]
+): void {
+    const entry = writes.unsent(key);
+    if (entry) {
+        writes.put({
+            ...entry,
+            value: replaySteps(entry.value, steps),
+            steps: [
+                ...(writes.isQueued(key) ? entry.steps : behind(entry)),
+                ...steps,
+            ],
+        });
+    }
+}

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

@@ -124,3 +124,23 @@ export function replaySteps(
     }
     return value;
 }
+
+/** The entries `next` changed against `read`, applied to `current`. */
+export function mergeEdits(
+    current: Record<string, unknown>,
+    read: Record<string, unknown>,
+    next: Record<string, unknown>
+): Record<string, unknown> {
+    const merged = { ...current };
+    for (const id of new Set([...Object.keys(read), ...Object.keys(next)])) {
+        if (JSON.stringify(read[id]) !== JSON.stringify(next[id])) {
+            const value = applyEdit(current[id], read[id], next[id]);
+            if (value === undefined) {
+                delete merged[id];
+            } else {
+                merged[id] = value;
+            }
+        }
+    }
+    return merged;
+}

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

@@ -35,6 +35,11 @@ export interface EditStep {
     readonly sent?: number;
 }
 
+/** A travelling write's steps, marked as carried by it. */
+export function behind(entry: PendingEntry): EditStep[] {
+    return entry.steps.map((step) => ({ ...step, sent: entry.edit }));
+}
+
 /** Steps whose write may or may not have landed. */
 export function uncertain(steps: readonly EditStep[]): EditStep[] {
     return steps.map((step) => ({ ...step, sent: 0 }));

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

@@ -8,6 +8,7 @@ import {
     entryKey,
     entryPath,
     addStep,
+    behind,
     type EditStep,
     type PendingEntry,
     preconditionHeaders,
@@ -46,7 +47,7 @@ export interface UserDataQueueHooks {
  */
 export class UserDataWriteQueue {
     /** Server version tag per entry; absent until the first read. */
-    private readonly tags = new UserStateTags();
+    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>();
@@ -109,13 +110,6 @@ export class UserDataWriteQueue {
         }
     }
 
-    setTags(
-        namespace: ServerUserStateNamespace,
-        tags: Readonly<Record<string, string>>
-    ): void {
-        this.tags.set(namespace, tags);
-    }
-
     /**
      * Queues an edit (`before` → `value`). A newer edit of an entry still
      * queued joins it; one behind a travelling write keeps that write's
@@ -146,17 +140,11 @@ export class UserDataWriteQueue {
                 steps: addStep(queued.steps, before, value),
             };
         } else if (travelling) {
-            const sent = travelling.steps.map((step) => ({
-                ...step,
-                sent: travelling.edit,
-            }));
             entry = {
-                namespace,
-                playlistId,
+                ...travelling,
                 value,
                 edit,
-                tag: travelling.tag,
-                steps: [...sent, { before, after: value }],
+                steps: [...behind(travelling), { before, after: value }],
             };
         } else {
             entry = {
@@ -180,45 +168,21 @@ 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,
-        current: unknown,
-        steps: readonly EditStep[],
-        foreign: boolean
-    ): unknown {
-        const key = `${namespace}|${playlistId}`;
-        const adopted = uncertain(steps);
-        const queued = this.pending.get(key);
-        if (queued) {
-            // This tab's queued edits are newer: they replay last.
-            const all = [...adopted, ...queued.steps];
-            const value = replaySteps(queued.steps[0]?.before, all);
-            const edit = ++this.editCount;
-            this.pending.set(key, {
-                ...queued,
-                value,
-                edit,
-                steps: all,
-                tag: foreign ? UNKNOWN_TAG : queued.tag,
-            });
-            this.edits.set(key, edit);
-            return value;
-        }
-        const value = replaySteps(current, steps);
-        this.queue(namespace, playlistId, value, value, { foreign });
-        const entry = this.pending.get(key);
-        if (entry) {
-            this.pending.set(key, {
-                ...entry,
-                steps: [...adopted, ...entry.steps],
-            });
-        }
-        return value;
+    /** The write of an entry: queued, else travelling. */
+    unsent(key: string): PendingEntry | undefined {
+        return this.pending.get(key) ?? this.inFlight.get(key);
+    }
+
+    isQueued(key: string): boolean {
+        return this.pending.has(key);
+    }
+
+    /** Queues `entry` as this tab's newest edit of it. */
+    put(entry: Omit<PendingEntry, 'edit'>): void {
+        const key = entryKey(entry as PendingEntry);
+        const edit = ++this.editCount;
+        this.pending.set(key, { ...entry, edit });
+        this.edits.set(key, edit);
     }
 
     scheduleFlush(): void {

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

@@ -12,10 +12,11 @@ import {
     splitEditKey,
     UNKNOWN_TAG,
 } from './server-user-data-entries';
+import { adoptEdits, extendUnsent } from './server-user-data-adopt';
 import { UserDataCache } from './server-user-data-cache';
 import {
-    applyEdit,
     type DeferredEdit,
+    mergeEdits,
     refreshedNamespace,
 } from './server-user-data-edit';
 import { UserDataWriteQueue } from './server-user-data-queue';
@@ -59,6 +60,8 @@ export class ServerUserDataStorage implements UserDataStorage {
     private hydrated = false;
     /** Closed tabs' page loads this one took over (once each). */
     private readonly adopted = new Set<string>();
+    /** Page loads taken over whose copy stays until this one saves. */
+    private readonly retained = new Set<string>();
     /** This page load: names its saved writes and its registry record. */
     private readonly tabId = Math.random().toString(36).slice(2);
     private readonly cache: UserDataCache;
@@ -129,9 +132,7 @@ export class ServerUserDataStorage implements UserDataStorage {
         // storage refuses the copy, the old one stays.
         const saved = [this.saveDeferred(), this.saveUnsent()].every(Boolean);
         this.publishUnconfirmed();
-        if (saved) {
-            removePage(this.local, this.instanceId, page);
-        }
+        this.release(page, saved);
     }
 
     /**
@@ -155,7 +156,8 @@ export class ServerUserDataStorage implements UserDataStorage {
             );
             for (const { namespace, playlistId, steps } of edits) {
                 // Queued even when the cache shows it already.
-                const value = this.writes.adopt(
+                const value = adoptEdits(
+                    this.writes,
                     namespace,
                     playlistId,
                     this.cache.map(namespace)[playlistId],
@@ -166,9 +168,7 @@ export class ServerUserDataStorage implements UserDataStorage {
             }
             this.adopted.add(page);
             // The closed tab's copy goes only once this tab saved its own.
-            if (this.saveUnsent()) {
-                removePage(this.local, this.instanceId, page);
-            }
+            this.release(page, this.saveUnsent());
             this.writes.scheduleFlush();
         }
     }
@@ -346,7 +346,7 @@ export class ServerUserDataStorage implements UserDataStorage {
                 base.set(
                     key,
                     own.has(key)
-                        ? cached[playlistId]
+                        ? this.writes.unsent(key)?.value
                         : entries[namespace]?.[playlistId]
                 );
             }
@@ -368,7 +368,7 @@ export class ServerUserDataStorage implements UserDataStorage {
                 for (const playlistId of keepCached) {
                     fresh[playlistId] = UNKNOWN_TAG;
                 }
-                this.writes.setTags(namespace, fresh);
+                this.writes.tags.set(namespace, fresh);
             }
         }
         this.writes.settledUpTo(editsBefore);
@@ -377,12 +377,17 @@ export class ServerUserDataStorage implements UserDataStorage {
             this.hydrated = true;
             this.deferred.clear();
             for (const edit of deferred) {
-                this.writes.queue(
-                    edit.namespace,
-                    edit.playlistId,
-                    refreshed.get(edit.namespace)?.[edit.playlistId],
-                    base.get(`${edit.namespace}|${edit.playlistId}`)
-                );
+                const key = `${edit.namespace}|${edit.playlistId}`;
+                if (own.has(key)) {
+                    extendUnsent(this.writes, key, edit.steps);
+                } else {
+                    this.writes.queue(
+                        edit.namespace,
+                        edit.playlistId,
+                        refreshed.get(edit.namespace)?.[edit.playlistId],
+                        base.get(key)
+                    );
+                }
             }
             this.saveDeferred();
             this.writes.scheduleFlush();
@@ -458,9 +463,25 @@ export class ServerUserDataStorage implements UserDataStorage {
         );
         this.unsentSaved = writes.length > 0 || !saved;
         this.publishUnconfirmed();
+        if (saved && this.retained.size > 0 && this.saveDeferred()) {
+            // This page's copy now stands for the ones it kept.
+            for (const page of this.retained) {
+                removePage(this.local, this.instanceId, page);
+            }
+            this.retained.clear();
+        }
         return saved;
     }
 
+    /** Forgets another page load's copy, or keeps it until this one saves. */
+    private release(page: string, saved: boolean): void {
+        if (saved) {
+            removePage(this.local, this.instanceId, page);
+        } else {
+            this.retained.add(page);
+        }
+    }
+
     private saveDeferred(): boolean {
         return saveDeferredEdits(this.local, this.instanceId, this.tabId, [
             ...this.deferred.values(),
@@ -474,23 +495,3 @@ export class ServerUserDataStorage implements UserDataStorage {
         ]);
     }
 }
-
-/** The entries `next` changed against `read`, applied to `current`. */
-function mergeEdits(
-    current: Record<string, unknown>,
-    read: Record<string, unknown>,
-    next: Record<string, unknown>
-): Record<string, unknown> {
-    const merged = { ...current };
-    for (const id of new Set([...Object.keys(read), ...Object.keys(next)])) {
-        if (JSON.stringify(read[id]) !== JSON.stringify(next[id])) {
-            const value = applyEdit(current[id], read[id], next[id]);
-            if (value === undefined) {
-                delete merged[id];
-            } else {
-                merged[id] = value;
-            }
-        }
-    }
-    return merged;
-}

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

@@ -811,6 +811,7 @@ describe('ServerUserDataStorage with other devices', () => {
             }
             setItem(key, value);
         };
+        const kept = saved()[0];
         const reloaded = new ServerUserDataStorage(
             offline as never,
             local,
@@ -820,7 +821,16 @@ describe('ServerUserDataStorage with other devices', () => {
         );
 
         expect(reloaded.hasUnsentWrites()).toBe(true);
+        expect(saved()).toEqual([kept]);
+
+        // Space again: this page's copy replaces the one it kept.
+        local.setItem = setItem;
+        reloaded.setItem(
+            'xtream-favorites',
+            JSON.stringify({ p1: [...favorites(reloaded).p1, 'Y'] })
+        );
         expect(saved()).toHaveLength(1);
+        expect(saved()).not.toEqual([kept]);
     });
 
     it('keeps a recovered write under an early edit made on it', async () => {
@@ -877,4 +887,52 @@ describe('ServerUserDataStorage with other devices', () => {
             [...(server.state['xtream-favorites']['p1'] as string[])].sort()
         ).toEqual(['S', 'new', 'old']);
     });
+
+    it('keeps an early removal of an item a recovered write adds', async () => {
+        const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } });
+        const local = memoryStorage();
+        const tabStore = memoryStorage();
+        const before = new ServerUserDataStorage(
+            {
+                ...server.api,
+                sendJson: jest
+                    .fn()
+                    .mockRejectedValue(
+                        new ServerApiError(0, 'network', 'offline')
+                    ),
+            } as never,
+            local,
+            jest.fn(),
+            'default',
+            tabStore
+        );
+        await before.hydrate();
+        before.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] }));
+        await before.flush();
+
+        const send = server.api.sendJson.getMockImplementation();
+        const reloaded = new ServerUserDataStorage(
+            {
+                ...server.api,
+                sendJson: jest
+                    .fn()
+                    .mockRejectedValueOnce(
+                        new ServerApiError(0, 'network', 'offline')
+                    )
+                    .mockImplementation(send),
+            } as never,
+            local,
+            jest.fn(),
+            'default',
+            tabStore
+        );
+        // X is taken back before the first read; another device adds Z.
+        reloaded.setItem('xtream-favorites', JSON.stringify({ p1: ['S'] }));
+        server.state['xtream-favorites']['p1'] = ['S', 'Z'];
+        await reloaded.hydrate();
+        await reloaded.flush();
+        await reloaded.flush();
+
+        expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'Z']);
+    });
 });