diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index 8a3b5ad91..e37857417 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -209,8 +209,10 @@ settings and guards read them in every build. re-reads run one at a time and keep entries edited while they were in flight or not yet saved. A re-read whose result does not fit the cache keeps the old cache and baseline together. A response body cut off after - its headers counts as offline. Before the first read finishes, an edit is - compared with the cache, so it sends only its own entry. A re-read + its headers counts as offline. Edits made before the first successful read + are held as before → after and applied to the server's value once it + arrives (`applyEdit`): a new device or a stale cache adds its favorite + instead of replacing the server's list. A re-read updates the cache; views already open show another device's changes on their next load (no live push). Start-up waits up to 3 s for the first read. diff --git a/libs/services/src/lib/server-storage-status.service.spec.ts b/libs/services/src/lib/server-storage-status.service.spec.ts index 10d241af6..71325d450 100644 --- a/libs/services/src/lib/server-storage-status.service.spec.ts +++ b/libs/services/src/lib/server-storage-status.service.spec.ts @@ -43,8 +43,9 @@ describe('ServerStorageStatusService', () => { status.reportUnauthorized(); expect(status.state()).toBe('unauthorized'); - // A late answer to a request sent with the old token. + // Late answers to requests sent with the old token. status.reportSuccess(); + status.reportOffline(); expect(status.state()).toBe('unauthorized'); }); }); diff --git a/libs/services/src/lib/server-storage-status.service.ts b/libs/services/src/lib/server-storage-status.service.ts index a1d8d49b4..ec36f91b2 100644 --- a/libs/services/src/lib/server-storage-status.service.ts +++ b/libs/services/src/lib/server-storage-status.service.ts @@ -41,7 +41,8 @@ export class ServerStorageStatusService { } reportOffline(): void { - if (this.isActive()) { + // A late failure after sign-out must not unlock the workspace. + if (this.isActive() && this.stateSignal() !== 'unauthorized') { this.stateSignal.set('offline'); } } diff --git a/libs/services/src/lib/server-storage/server-migration-merge.ts b/libs/services/src/lib/server-storage/server-migration-merge.ts index 9d4266f5e..b5197740b 100644 --- a/libs/services/src/lib/server-storage/server-migration-merge.ts +++ b/libs/services/src/lib/server-storage/server-migration-merge.ts @@ -45,7 +45,7 @@ export function unionItems(...lists: readonly (readonly T[])[]): T[] { * Content ids or URLs identify favorites and history entries; a playback * position is identified by its content type and Xtream id. */ -function itemKey(item: unknown): string { +export function itemKey(item: unknown): string { if (!isPlainObject(item)) { return `value:${JSON.stringify(item)}`; } diff --git a/libs/services/src/lib/server-storage/server-unsent-writes.ts b/libs/services/src/lib/server-storage/server-unsent-writes.ts index 33939b6f2..7c54c9c76 100644 --- a/libs/services/src/lib/server-storage/server-unsent-writes.ts +++ b/libs/services/src/lib/server-storage/server-unsent-writes.ts @@ -104,3 +104,18 @@ export function saveUnsentWrites( // Storage unavailable or full: the queue stays in memory only. } } + +/** One write per entry, the newest by edit number. */ +export function newestWrites( + writes: readonly T[] +): T[] { + const newest = new Map(); + for (const write of writes) { + const key = `${write.namespace}|${write.playlistId}`; + const known = newest.get(key); + if (!known || write.edit > known.edit) { + newest.set(key, write); + } + } + return [...newest.values()]; +} diff --git a/libs/services/src/lib/server-storage/server-user-data-edit.spec.ts b/libs/services/src/lib/server-storage/server-user-data-edit.spec.ts new file mode 100644 index 000000000..53353fff3 --- /dev/null +++ b/libs/services/src/lib/server-storage/server-user-data-edit.spec.ts @@ -0,0 +1,61 @@ +import { applyEdit } from './server-user-data-edit'; + +describe('applyEdit', () => { + it('adds the new favorite to what the server already has', () => { + expect(applyEdit(['live:1', 'live:2'], [], ['live:9'])).toEqual([ + 'live:1', + 'live:2', + 'live:9', + ]); + expect(applyEdit(['live:1'], undefined, ['live:9'])).toEqual([ + 'live:1', + 'live:9', + ]); + }); + + it('removes only what the edit removed and replaces changed items', () => { + expect( + applyEdit( + [ + { + contentType: 'vod', + contentXtreamId: 1, + positionSeconds: 5, + }, + 'other', + ], + [ + { + contentType: 'vod', + contentXtreamId: 1, + positionSeconds: 5, + }, + ], + [ + { + contentType: 'vod', + contentXtreamId: 1, + positionSeconds: 60, + }, + ] + ) + ).toEqual([ + { contentType: 'vod', contentXtreamId: 1, positionSeconds: 60 }, + 'other', + ]); + expect(applyEdit(['a', 'b'], ['a'], [])).toEqual(['b']); + // Another device removed 'a'; this edit only added 'c'. + expect(applyEdit(['b'], ['a', 'b'], ['a', 'b', 'c'])).toEqual([ + 'b', + 'c', + ]); + }); + + it('merges maps by key and passes deletes through', () => { + expect(applyEdit({ a: 1, b: 2 }, { a: 1 }, { c: 3 })).toEqual({ + b: 2, + c: 3, + }); + expect(applyEdit(['x'], ['x'], undefined)).toBeUndefined(); + }); +}); 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 new file mode 100644 index 000000000..edd45a6bf --- /dev/null +++ b/libs/services/src/lib/server-storage/server-user-data-edit.ts @@ -0,0 +1,116 @@ +import type { ServerUserStateNamespace } from '@iptvnator/shared/interfaces'; +import { itemKey } from './server-migration-merge'; + +/** An edit made before the first read, applied once the server's value is known. */ +export interface DeferredEdit { + readonly namespace: ServerUserStateNamespace; + readonly playlistId: string; + readonly before: unknown; + readonly after: unknown; +} + +/** + * One namespace's cache after a read: the server's map, except entries the + * server does not have yet (`keepCached`, from the cache) and edits made + * before the first read (applied to the server's value). + */ +export function refreshedNamespace( + server: Readonly>, + cached: Readonly>, + keepCached: ReadonlySet, + deferred: readonly DeferredEdit[] +): Record { + const value: Record = { ...server }; + for (const playlistId of keepCached) { + if (playlistId in cached) { + value[playlistId] = cached[playlistId]; + } else { + delete value[playlistId]; + } + } + for (const edit of deferred) { + const merged = applyEdit( + server[edit.playlistId], + edit.before, + edit.after + ); + if (merged === undefined) { + delete value[edit.playlistId]; + } else { + value[edit.playlistId] = merged; + } + } + return value; +} + +/** + * Applies one local edit (`before` → `after`, as the user made it against + * an older or empty cache) to the server's current value: list items the + * edit added or changed are set, items it removed are dropped, everything + * else the server has stays. Maps work the same by key. + */ +export function applyEdit( + server: unknown, + before: unknown, + after: unknown +): unknown { + if (after === undefined) { + return undefined; + } + if ( + Array.isArray(after) && + isListOrMissing(server) && + isListOrMissing(before) + ) { + const was = new Map( + (before ?? []).map((item) => [itemKey(item), item]) + ); + const now = new Map(after.map((item) => [itemKey(item), item])); + const result = (server ?? []).filter((item) => { + const key = itemKey(item); + return !(was.has(key) && !now.has(key)); + }); + for (const [key, item] of now) { + if (JSON.stringify(was.get(key)) === JSON.stringify(item)) { + // Untouched by the edit: the server's state of it wins. + continue; + } + const index = result.findIndex((entry) => itemKey(entry) === key); + if (index < 0) { + result.push(item); + } else { + result[index] = item; + } + } + return result; + } + if (isMap(after) && isMapOrMissing(server) && isMapOrMissing(before)) { + const result: Record = { ...(server ?? {}) }; + for (const key of Object.keys(before ?? {})) { + if (!(key in after)) { + delete result[key]; + } + } + for (const [key, value] of Object.entries(after)) { + if (JSON.stringify((before ?? {})[key]) !== JSON.stringify(value)) { + result[key] = value; + } + } + return result; + } + return after; +} + +function isMap(value: unknown): value is Record { + return !!value && typeof value === 'object' && !Array.isArray(value); +} + +function isMapOrMissing( + value: unknown +): value is Record | undefined { + return value === undefined || isMap(value); +} + +function isListOrMissing(value: unknown): value is unknown[] | undefined { + return value === undefined || Array.isArray(value); +} diff --git a/libs/services/src/lib/server-storage/server-user-data-storage.spec.ts b/libs/services/src/lib/server-storage/server-user-data-storage.spec.ts index dff3ec3c1..8f6499a22 100644 --- a/libs/services/src/lib/server-storage/server-user-data-storage.spec.ts +++ b/libs/services/src/lib/server-storage/server-user-data-storage.spec.ts @@ -325,7 +325,7 @@ describe('ServerUserDataStorage', () => { ); }); - it('sends only the edited entry before the first read finished', () => { + it('applies an edit made before the first read to the server value', async () => { const { api, local, storage } = setup(); // The cache from the last visit; no read from the server yet. local.setItem( @@ -337,17 +337,51 @@ describe('ServerUserDataStorage', () => { 'xtream-favorites', JSON.stringify({ p1: ['a', 'c'], p2: ['b'] }) ); - void storage.flush(); + await storage.flush(); + expect(api.sendJson).not.toHaveBeenCalled(); - return settle().then(() => { - expect(api.sendJson.mock.calls).toEqual([ - [ - 'PUT', - '/user-state/xtream-favorites/p1', - { value: ['a', 'c'] }, - ], - ]); + // Meanwhile another device added 'live:1' and 'z'. + api.getJson.mockResolvedValueOnce({ + entries: { + 'xtream-favorites': { p1: ['live:1', 'a'], p2: ['b', 'z'] }, + }, }); + await storage.hydrate(); + await storage.flush(); + + expect(api.sendJson.mock.calls).toEqual([ + [ + 'PUT', + '/user-state/xtream-favorites/p1', + { value: ['live:1', 'a', 'c'] }, + ], + ]); + expect(JSON.parse(storage.getItem('xtream-favorites') ?? '')).toEqual({ + p1: ['live:1', 'a', 'c'], + p2: ['b', 'z'], + }); + }); + + it('keeps the server favorites when a new device adds one early', async () => { + const { api, storage } = setup(); + let answer: (value: unknown) => void = () => undefined; + api.getJson.mockReturnValueOnce( + new Promise((resolve) => (answer = resolve)) + ); + + const firstRead = storage.hydrate(); + storage.setItem('xtream-favorites', JSON.stringify({ p1: ['live:9'] })); + answer({ + entries: { 'xtream-favorites': { p1: ['live:1', 'live:2'] } }, + }); + await firstRead; + await storage.flush(); + + expect(api.sendJson).toHaveBeenCalledWith( + 'PUT', + '/user-state/xtream-favorites/p1', + { value: ['live:1', 'live:2', 'live:9'] } + ); }); it('never retries an older page-hide write over a newer one', async () => { @@ -506,4 +540,33 @@ describe('ServerUserDataStorage', () => { ); expect(local.values.get('iptvnator-server:cache-owner')).toBe('new-db'); }); + + it('keeps a newer travelling write when an older one is answered', async () => { + const { api, local, storage } = setup(); + await storage.hydrate(); + let answerOlder: () => void = () => undefined; + api.sendJson.mockReturnValueOnce( + new Promise((resolve) => (answerOlder = resolve)) + ); + api.sendKeepalive.mockReturnValueOnce(new Promise(() => undefined)); + + storage.setItem('xtream-favorites', JSON.stringify({ p1: ['old'] })); + const older = storage.flush(); + await settle(); + storage.setItem('xtream-favorites', JSON.stringify({ p1: ['new'] })); + storage.flushWithKeepalive(); + answerOlder(); + await older; + await settle(); + + expect(storage.hasUnsentWrites()).toBe(true); + expect( + JSON.parse( + local.values.get('iptvnator-server:unsent-writes:default') ?? + '[]' + ) + ).toEqual([ + { namespace: 'xtream-favorites', playlistId: 'p1', value: ['new'] }, + ]); + }); }); 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 26c99c60d..ba8585eeb 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 @@ -13,8 +13,10 @@ import { type PendingEntry, splitEditKey, } from './server-user-data-entries'; +import { type DeferredEdit, refreshedNamespace } from './server-user-data-edit'; import { claimCache, + newestWrites, readUnsentWrites, saveUnsentWrites, } from './server-unsent-writes'; @@ -47,6 +49,13 @@ export class ServerUserDataStorage implements UserDataStorage { /** Writes still travelling (normal or page-hide), by entry key. */ private readonly inFlight = new Map(); private unsentSaved = false; + /** + * Edits made before the first successful read, as before → after: the + * cache they were made against may be empty or old, so they are + * applied to the server's value once it is known (`applyEdit`). + */ + private readonly deferred = new Map(); + private hydrated = false; /** Re-reads run one at a time, so an older one never lands last. */ private hydration: Promise = Promise.resolve(); /** Edit counter per `namespace|playlistId`, for hydrate to keep them. */ @@ -71,7 +80,11 @@ 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; + return ( + this.pending.size > 0 || + this.inFlight.size > 0 || + this.deferred.size > 0 + ); } getItem(key: string): string | null { @@ -120,33 +133,32 @@ export class ServerUserDataStorage implements UserDataStorage { await this.flush(); const { entries } = await this.api.getJson('/user-state'); - for (const namespace of SERVER_USER_STATE_NAMESPACES) { - const value: Record = { - ...(entries[namespace] ?? {}), - }; - const current = parseMap( - this.local.getItem(SERVER_CACHE_PREFIX + namespace) ?? '{}' - ); - const unsaved = new Set([ - ...this.pending.keys(), - ...this.inFlight.keys(), - ]); - for (const [key, edit] of this.edits) { - if (edit > editsBefore) { - unsaved.add(key); - } + const unsaved = new Set([ + ...this.pending.keys(), + ...this.inFlight.keys(), + ]); + for (const [key, edit] of this.edits) { + if (edit > editsBefore) { + unsaved.add(key); } + } + const deferred = [...this.deferred.values()]; + for (const namespace of SERVER_USER_STATE_NAMESPACES) { + const keepCached = new Set(); for (const key of unsaved) { const [editedNamespace, playlistId] = splitEditKey(key); - if (editedNamespace !== namespace) { - continue; - } - if (playlistId in current) { - value[playlistId] = current[playlistId]; - } else { - delete value[playlistId]; + if (editedNamespace === namespace) { + keepCached.add(playlistId); } } + const value = refreshedNamespace( + entries[namespace] ?? {}, + parseMap( + this.local.getItem(SERVER_CACHE_PREFIX + namespace) ?? '{}' + ), + keepCached, + deferred.filter((edit) => edit.namespace === namespace) + ); try { this.local.setItem( SERVER_CACHE_PREFIX + namespace, @@ -168,6 +180,19 @@ export class ServerUserDataStorage implements UserDataStorage { this.edits.delete(key); } } + if (!this.hydrated) { + // From here edits go straight to the queue. + this.hydrated = true; + this.deferred.clear(); + for (const edit of deferred) { + this.queue( + edit.namespace, + edit.playlistId, + this.synced.get(edit.namespace)?.[edit.playlistId] + ); + } + this.scheduleFlush(); + } } /** Sends the queued entries after any batch still travelling. */ @@ -253,16 +278,29 @@ export class ServerUserDataStorage implements UserDataStorage { next: Record, previous: Record ): void { + const change = (playlistId: string, value: unknown) => { + if (this.hydrated) { + this.queue(namespace, playlistId, value); + return; + } + const key = `${namespace}|${playlistId}`; + this.deferred.set(key, { + namespace, + playlistId, + before: this.deferred.get(key)?.before ?? previous[playlistId], + after: value, + }); + }; for (const [playlistId, value] of Object.entries(next)) { if ( JSON.stringify(previous[playlistId]) !== JSON.stringify(value) ) { - this.queue(namespace, playlistId, value); + change(playlistId, value); } } for (const playlistId of Object.keys(previous)) { if (!(playlistId in next)) { - this.queue(namespace, playlistId, undefined); + change(playlistId, undefined); } } this.synced.set(namespace, next); @@ -320,7 +358,7 @@ export class ServerUserDataStorage implements UserDataStorage { const ok = await (entry.value === undefined ? this.api.sendKeepalive('DELETE', path) : this.api.sendKeepalive('PUT', path, { value: entry.value })); - this.inFlight.delete(entryKey(entry)); + this.retireInFlight(entry); if (!ok) { this.requeue(entry); } @@ -336,19 +374,31 @@ export class ServerUserDataStorage implements UserDataStorage { value: entry.value, }); } - this.inFlight.delete(entryKey(entry)); + this.retireInFlight(entry); this.saveUnsent(); } catch (error) { - this.inFlight.delete(entryKey(entry)); + 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); + } + } + /** Persists what the server has not confirmed; skips idle rewrites. */ private saveUnsent(): void { - const writes = [...this.pending.values(), ...this.inFlight.values()]; + // A newer edit can be queued while an older one is still travelling. + const writes = newestWrites([ + ...this.inFlight.values(), + ...this.pending.values(), + ]); if (writes.length === 0 && !this.unsentSaved) { return; }