From 716580cb16c4519a4036cf79d722c2751fd305fa Mon Sep 17 00:00:00 2001 From: 4gray Date: Mon, 12 Oct 2026 23:30:31 +0200 Subject: [PATCH] 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 --- docs/architecture/pwa-self-hosted.md | 12 ++- libs/m3u-state/src/lib/effects.ts | 18 +++- .../src/lib/favorites.effect.spec.ts | 93 +++++++++++++++++++ .../server-storage/server-user-data-adopt.ts | 68 ++++++++++++++ .../server-storage/server-user-data-edit.ts | 20 ++++ .../server-user-data-entries.ts | 5 + .../server-storage/server-user-data-queue.ts | 74 ++++----------- .../server-user-data-storage.ts | 73 ++++++++------- .../server-user-data-sync.spec.ts | 58 ++++++++++++ 9 files changed, 322 insertions(+), 99 deletions(-) create mode 100644 libs/m3u-state/src/lib/favorites.effect.spec.ts create mode 100644 libs/services/src/lib/server-storage/server-user-data-adopt.ts diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index add1e69b0..009ea862e 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/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::`). 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::`). A shared registry in `localStorage` (`iptvnator-server:unconfirmed:`, one record per page load, renewed every 30 s while it has unsent writes, diff --git a/libs/m3u-state/src/lib/effects.ts b/libs/m3u-state/src/lib/effects.ts index 8615b1f57..2dc688530 100644 --- a/libs/m3u-state/src/lib/effects.ts +++ b/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'), diff --git a/libs/m3u-state/src/lib/favorites.effect.spec.ts b/libs/m3u-state/src/lib/favorites.effect.spec.ts new file mode 100644 index 000000000..4a1a67bb5 --- /dev/null +++ b/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; + 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']); + }); +}); diff --git a/libs/services/src/lib/server-storage/server-user-data-adopt.ts b/libs/services/src/lib/server-storage/server-user-data-adopt.ts new file mode 100644 index 000000000..e79c96893 --- /dev/null +++ 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, + ], + }); + } +} diff --git a/libs/services/src/lib/server-storage/server-user-data-edit.ts b/libs/services/src/lib/server-storage/server-user-data-edit.ts index cebd2a709..c17812c82 100644 --- a/libs/services/src/lib/server-storage/server-user-data-edit.ts +++ b/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, + read: Record, + next: Record +): Record { + 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; +} diff --git a/libs/services/src/lib/server-storage/server-user-data-entries.ts b/libs/services/src/lib/server-storage/server-user-data-entries.ts index de9b323fd..edd85171f 100644 --- a/libs/services/src/lib/server-storage/server-user-data-entries.ts +++ b/libs/services/src/lib/server-storage/server-user-data-entries.ts @@ -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 })); diff --git a/libs/services/src/lib/server-storage/server-user-data-queue.ts b/libs/services/src/lib/server-storage/server-user-data-queue.ts index 5e2aba41e..f6edd0c70 100644 --- a/libs/services/src/lib/server-storage/server-user-data-queue.ts +++ b/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(); /** Writes still travelling (normal or page-hide), by entry key. */ private readonly inFlight = new Map(); @@ -109,13 +110,6 @@ export class UserDataWriteQueue { } } - setTags( - namespace: ServerUserStateNamespace, - tags: Readonly> - ): 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): void { + const key = entryKey(entry as PendingEntry); + const edit = ++this.editCount; + this.pending.set(key, { ...entry, edit }); + this.edits.set(key, edit); } scheduleFlush(): void { diff --git a/libs/services/src/lib/server-storage/server-user-data-storage.ts b/libs/services/src/lib/server-storage/server-user-data-storage.ts index ad6411a52..dd0cfbfe4 100644 --- a/libs/services/src/lib/server-storage/server-user-data-storage.ts +++ b/libs/services/src/lib/server-storage/server-user-data-storage.ts @@ -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(); + /** Page loads taken over whose copy stays until this one saves. */ + private readonly retained = new Set(); /** 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, - read: Record, - next: Record -): Record { - 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; -} diff --git a/libs/services/src/lib/server-storage/server-user-data-sync.spec.ts b/libs/services/src/lib/server-storage/server-user-data-sync.spec.ts index fd1564523..1c7632a58 100644 --- a/libs/services/src/lib/server-storage/server-user-data-sync.spec.ts +++ b/libs/services/src/lib/server-storage/server-user-data-sync.spec.ts @@ -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']); + }); });