diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index e89adb84e..f7b128ea4 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -206,9 +206,12 @@ settings and guards read them in every build. `keepalive` flush when the page hides. Batches go out one after another (the page-hide flush waits for a batch still travelling), a failed write, including a refused `keepalive` one, stays queued for the next flush, and - every unconfirmed write (queued or travelling) is saved in the tab's own - `sessionStorage` (`iptvnator-server:unsent-writes:`) and sent - first after a reload; tabs never clear each other's. + every unconfirmed write (queued or travelling) is saved in `localStorage` + 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 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. Cache and saved writes belong to one server database (`iptvnator-server:cache-owner`); another one behind the same address starts clean. Also, @@ -221,19 +224,21 @@ settings and guards read them in every build. keeps its own before → after and carries the travelling one as `prior` (also in the saved writes): if that write was lost, a conflict replays both in order, and if it landed, adding and then removing an item still - removes it. An entry a re-read keeps from the cache gets no tag, so its - next write conflicts and is replayed too. Page-hide writes report their + removes it. An edit made on another tab's unsent change, and an entry a + re-read keeps from the cache, are sent on a version no entry has + (`UNKNOWN_TAG`), so they conflict and are replayed too. Page-hide writes report their new tag or conflict like any other, and one behind a travelling write is taken from the queue only when that one is answered. While a playlist is 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`): a new device or a stale cache adds its favorite - instead of replacing the server's list. They are saved in the tab too - (`iptvnator-server:deferred-edits:`). A shared registry in - `localStorage` (`iptvnator-server:unconfirmed:`, one record - per tab, ignored after ten minutes) lets a re-read in one tab keep the - cache values another tab is still sending. A value too big for + instead of replacing the server's list. 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, + stale after ten minutes) lets a re-read in one tab keep the cache values + another tab is still sending. A value too big for `localStorage` stays readable from memory, so callers and the diff baseline agree. A re-read updates the cache; views already open show another device's changes on @@ -272,7 +277,8 @@ settings and guards read them in every build. lists (playback positions by content type and Xtream id, renamed to the server playlist) and merges maps, the server's value winning on a conflict. Parental locks stay in the browser but move to the server - playlist id with it. Browser data is never + playlist id with it; while the lock store cannot be read the move counts + as failed. Browser data is never deleted, the move does not start while user-data writes are unsent, and the marker is written only when nothing failed, so Settings → Server keeps offering a partial move. diff --git a/libs/services/src/lib/server-storage/server-migration.service.spec.ts b/libs/services/src/lib/server-storage/server-migration.service.spec.ts index f56b7282c..d62011624 100644 --- a/libs/services/src/lib/server-storage/server-migration.service.spec.ts +++ b/libs/services/src/lib/server-storage/server-migration.service.spec.ts @@ -33,14 +33,14 @@ describe('ServerMigrationService', () => { const emptyLocks = () => ({ xtream: [], stalker: [], m3u: [] }); let parentalLock: { - initialize: jest.Mock; + ensureLocksReadable: jest.Mock; locksFor: jest.Mock; replacePlaylistLocks: jest.Mock; }; beforeEach(() => { parentalLock = { - initialize: jest.fn().mockResolvedValue(undefined), + ensureLocksReadable: jest.fn().mockResolvedValue(true), locksFor: jest.fn(() => emptyLocks()), replacePlaylistLocks: jest.fn().mockResolvedValue(true), }; @@ -367,4 +367,23 @@ describe('ServerMigrationService', () => { } ); }); + + it('keeps the move on offer while parental locks cannot be read', async () => { + session.documentStore.dbGetAppPlaylistMetas.mockResolvedValue([ + { + _id: 'server-x', + title: 'Provider', + serverUrl: 'http://panel.test', + username: 'user', + }, + ]); + parentalLock.ensureLocksReadable.mockResolvedValue(false); + + const result = await TestBed.inject( + ServerMigrationService + ).moveBrowserData(); + + expect(result.failed).toBeGreaterThan(0); + expect(parentalLock.replacePlaylistLocks).not.toHaveBeenCalled(); + }); }); diff --git a/libs/services/src/lib/server-storage/server-migration.service.ts b/libs/services/src/lib/server-storage/server-migration.service.ts index 4791ed4ef..f8ff50802 100644 --- a/libs/services/src/lib/server-storage/server-migration.service.ts +++ b/libs/services/src/lib/server-storage/server-migration.service.ts @@ -279,7 +279,14 @@ export class ServerMigrationService { private async moveParentalLocks( targets: ReadonlyMap ): Promise { - await this.parentalLock.initialize(); + const moving = [...targets].filter(([from, to]) => from !== to); + // An unreadable lock store reads as empty: keep the move on offer. + if ( + moving.length > 0 && + !(await this.parentalLock.ensureLocksReadable()) + ) { + return moving.length; + } let failed = 0; for (const [browserId, serverId] of targets) { const from = this.parentalLock.locksFor(browserId); diff --git a/libs/services/src/lib/server-storage/server-storage-session.spec.ts b/libs/services/src/lib/server-storage/server-storage-session.spec.ts index 46e6b3a38..44fb7889f 100644 --- a/libs/services/src/lib/server-storage/server-storage-session.spec.ts +++ b/libs/services/src/lib/server-storage/server-storage-session.spec.ts @@ -128,9 +128,10 @@ describe('ServerStorageSession', () => { }); it('sends no write before the reload checks the instance', async () => { - // A write the last visit could not send, for instance i-1. - sessionStorage.setItem( - 'iptvnator-server:unsent-writes:i-1', + // A write this tab's last page load could not send. + sessionStorage.setItem('iptvnator-server:page:i-1', 'last'); + localStorage.setItem( + 'iptvnator-server:unsent-writes:i-1:last', JSON.stringify([ { namespace: 'xtream-favorites', diff --git a/libs/services/src/lib/server-storage/server-storage-session.ts b/libs/services/src/lib/server-storage/server-storage-session.ts index 14ef3d5ca..f71c9095e 100644 --- a/libs/services/src/lib/server-storage/server-storage-session.ts +++ b/libs/services/src/lib/server-storage/server-storage-session.ts @@ -139,6 +139,7 @@ export class ServerStorageSession { this.userData.flushWithKeepalive() ); setInterval(() => { + this.userData.touch(); if (this.status.state() === 'offline') { void this.refresh(); } 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 c5063fad9..2195c75cf 100644 --- a/libs/services/src/lib/server-storage/server-unsent-writes.ts +++ b/libs/services/src/lib/server-storage/server-unsent-writes.ts @@ -6,20 +6,25 @@ import { import type { UserDataStorage } from '../user-data-storage'; /** - * User-data writes the server has not confirmed, kept across reloads: a - * failed write, or a page-hide write whose answer the page did not live to - * see, is sent again on the next visit instead of being replaced by the - * server's older value. + * User-data writes the server has not confirmed, kept in `localStorage` + * per page load: a failed write, or a page-hide write whose answer the page + * did not live to see, is sent again after a reload (by the same tab) or + * after the tab closed (by another, see `orphanedPages`), instead of being + * replaced by the server's older value. */ export const UNSENT_WRITES_KEY = 'iptvnator-server:unsent-writes'; /** Which server database the cached user data belongs to. */ export const CACHE_OWNER_KEY = 'iptvnator-server:cache-owner'; +/** The tab's current page load, in the tab's own storage. */ +export const PAGE_KEY = 'iptvnator-server:page'; -const writesKey = (instanceId: string) => `${UNSENT_WRITES_KEY}:${instanceId}`; +const writesKey = (instanceId: string, page: string) => + `${UNSENT_WRITES_KEY}:${instanceId}:${page}`; /** - * Which entries each open tab has not had confirmed yet, so a re-read in - * one tab keeps the cache values another tab is still sending. A record - * older than `UNCONFIRMED_TTL_MS` belongs to a tab that is gone. + * Which entries each page load has not had confirmed yet, so a re-read in + * one tab keeps the cache values another tab is still sending. A live tab + * renews its record; one older than `UNCONFIRMED_TTL_MS` belongs to a tab + * that is gone. The records also list whose saved writes exist. */ export const UNCONFIRMED_KEY = 'iptvnator-server:unconfirmed'; export const UNCONFIRMED_TTL_MS = 10 * 60 * 1000; @@ -28,8 +33,8 @@ const unconfirmedKey = (instanceId: string) => /** Edits made before the first read of a visit, kept as before → after. */ export const DEFERRED_EDITS_KEY = 'iptvnator-server:deferred-edits'; -const deferredKey = (instanceId: string) => - `${DEFERRED_EDITS_KEY}:${instanceId}`; +const deferredKey = (instanceId: string, page: string) => + `${DEFERRED_EDITS_KEY}:${instanceId}:${page}`; /** * Another database behind this address (a new volume, another backend): @@ -46,9 +51,11 @@ export function claimCache(local: UserDataStorage, instanceId: string): void { local.removeItem(`iptvnator-server:${namespace}`); } if (owner) { + for (const page of Object.keys(readRegistry(local, owner))) { + local.removeItem(writesKey(owner, page)); + local.removeItem(deferredKey(owner, page)); + } local.removeItem(unconfirmedKey(owner)); - local.removeItem(writesKey(owner)); - local.removeItem(deferredKey(owner)); } local.setItem(CACHE_OWNER_KEY, instanceId); } catch { @@ -80,13 +87,63 @@ interface StoredWrite { readonly prior?: { readonly before?: unknown; readonly after?: unknown }; } +/** + * Records `page` as the tab's current page load and returns the one + * before it: a reload, whose saved writes this page takes over. + */ +export function swapPage( + tabStore: UserDataStorage, + instanceId: string, + page: string +): string | null { + try { + const previous = tabStore.getItem(`${PAGE_KEY}:${instanceId}`); + tabStore.setItem(`${PAGE_KEY}:${instanceId}`, page); + return previous; + } catch { + return null; + } +} + +/** Page loads whose record went stale: closed tabs nobody sends for. */ +export function orphanedPages( + local: UserDataStorage, + instanceId: string, + page: string, + now = Date.now() +): string[] { + return Object.entries(readRegistry(local, instanceId)) + .filter( + ([owner, record]) => + owner !== page && + !(now - Number(record?.at) < UNCONFIRMED_TTL_MS) + ) + .map(([owner]) => owner); +} + +/** Forgets a page load's saved writes, early edits and record. */ +export function removePage( + local: UserDataStorage, + instanceId: string, + page: string +): void { + try { + local.removeItem(writesKey(instanceId, page)); + local.removeItem(deferredKey(instanceId, page)); + } catch { + // Storage unavailable: nothing was saved there either. + } + markUnconfirmed(local, instanceId, page, []); +} + export function readUnsentWrites( local: UserDataStorage, - instanceId: string + instanceId: string, + page: string ): UnsentWrite[] { let stored: unknown; try { - stored = JSON.parse(local.getItem(writesKey(instanceId)) ?? '[]'); + stored = JSON.parse(local.getItem(writesKey(instanceId, page)) ?? '[]'); } catch { return []; } @@ -121,11 +178,12 @@ export function readUnsentWrites( export function saveUnsentWrites( local: UserDataStorage, instanceId: string, + page: string, writes: readonly UnsentWrite[] ): void { try { if (writes.length === 0) { - local.removeItem(writesKey(instanceId)); + local.removeItem(writesKey(instanceId, page)); return; } const stored: StoredWrite[] = writes.map( @@ -140,7 +198,7 @@ export function saveUnsentWrites( : { prior: { before: prior.before, after: prior.after } }), }) ); - local.setItem(writesKey(instanceId), JSON.stringify(stored)); + local.setItem(writesKey(instanceId, page), JSON.stringify(stored)); } catch { // Storage unavailable or full: the queue stays in memory only. } @@ -170,11 +228,14 @@ export interface SavedDeferredEdit { export function readDeferredEdits( local: UserDataStorage, - instanceId: string + instanceId: string, + page: string ): SavedDeferredEdit[] { let stored: unknown; try { - stored = JSON.parse(local.getItem(deferredKey(instanceId)) ?? '[]'); + stored = JSON.parse( + local.getItem(deferredKey(instanceId, page)) ?? '[]' + ); } catch { return []; } @@ -204,15 +265,16 @@ export function readDeferredEdits( export function saveDeferredEdits( local: UserDataStorage, instanceId: string, + page: string, edits: readonly SavedDeferredEdit[] ): void { try { if (edits.length === 0) { - local.removeItem(deferredKey(instanceId)); + local.removeItem(deferredKey(instanceId, page)); return; } local.setItem( - deferredKey(instanceId), + deferredKey(instanceId, page), JSON.stringify( edits.map((edit) => ({ namespace: edit.namespace, 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 edd45a6bf..2f0c52ab3 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 @@ -114,3 +114,18 @@ function isMapOrMissing( function isListOrMissing(value: unknown): value is unknown[] | undefined { return value === undefined || Array.isArray(value); } + +/** A saved write's edits, its `prior` first, applied to `server`. */ +export function replayWrite( + server: unknown, + write: { + readonly before?: unknown; + readonly value: unknown; + readonly prior?: { readonly before: unknown; readonly after: unknown }; + } +): unknown { + const base = write.prior + ? applyEdit(server, write.prior.before, write.prior.after) + : server; + return applyEdit(base, write.before, write.value); +} 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 1cd75e875..cc088c40d 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 @@ -30,6 +30,13 @@ export interface PriorEdit { readonly edit: number; } +/** + * A version no entry ever has: a write based on what the cache shows but + * the server may not have (another tab's unsent change) always conflicts + * and is replayed on the server's value. + */ +export const UNKNOWN_TAG = '~unknown'; + /** `If-Match` / `If-None-Match` for a write based on `tag`. */ export function preconditionHeaders( tag: string | null | undefined 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 30a0f7ed9..f31731bdf 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 @@ -10,8 +10,9 @@ import { type PendingEntry, preconditionHeaders, type PriorEdit, + UNKNOWN_TAG, } from './server-user-data-entries'; -import { applyEdit } from './server-user-data-edit'; +import { replayWrite } from './server-user-data-edit'; import { newestWrites } from './server-unsent-writes'; const FLUSH_DELAY_MS = 500; @@ -118,15 +119,20 @@ export class UserDataWriteQueue { /** * Queues an edit (`before` → `value`). A newer edit of an entry still * queued joins it; one behind a travelling write keeps its own edit, - * carries the write's as `prior` and is based on the write's tag. + * carries the write's as `prior` and is based on the write's tag. A + * `foreign` edit was made on another tab's unsent change: its write + * must conflict and be replayed. */ queue( namespace: ServerUserStateNamespace, playlistId: string, value: unknown, before: unknown, - tag = this.tagFor(namespace, playlistId), - prior?: Omit + base: { + readonly tag?: string | null; + readonly prior?: Omit; + readonly foreign?: boolean; + } = {} ): void { const key = `${namespace}|${playlistId}`; const queued = this.pending.get(key); @@ -152,11 +158,18 @@ export class UserDataWriteQueue { }, }; } else { + const tag = + base.tag !== undefined + ? base.tag + : this.tagFor(namespace, playlistId); entry = { namespace, playlistId, value, edit, before, tag }; - if (prior) { - entry = { ...entry, prior: { ...prior, edit: 0 } }; + if (base.prior) { + entry = { ...entry, prior: { ...base.prior, edit: 0 } }; } } + if (base.foreign) { + entry = { ...entry, tag: UNKNOWN_TAG }; + } this.pending.set(key, entry); this.edits.set(key, edit); } @@ -410,7 +423,12 @@ export class UserDataWriteQueue { const key = entryKey(entry); const next = this.pending.get(key); if (next?.prior?.edit === entry.edit) { - this.pending.set(key, { ...next, prior: undefined, tag }); + // A foreign edit stays unknown: its baseline is not this value. + this.pending.set(key, { + ...next, + prior: undefined, + tag: next.tag === UNKNOWN_TAG ? UNKNOWN_TAG : tag, + }); } } @@ -432,7 +450,7 @@ export class UserDataWriteQueue { if (!latest) { return; } - const merged = replay(current.value, latest); + const merged = replayWrite(current.value, latest); this.hooks.show(entry.namespace, entry.playlistId, merged); if ( latest === entry && @@ -450,11 +468,3 @@ export class UserDataWriteQueue { this.scheduleFlush(); } } - -/** An entry's edits, its `prior` first, applied to the server's value. */ -function replay(server: unknown, entry: PendingEntry): unknown { - const base = entry.prior - ? applyEdit(server, entry.prior.before, entry.prior.after) - : server; - return applyEdit(base, entry.before, entry.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 0b7266e10..ef48b93aa 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 @@ -12,6 +12,18 @@ function memoryStorage() { } /** Lets chained promises run without advancing the fake timers. */ +/** The saved writes of the one page load that has any, for `instanceId`. */ +function savedWrites( + local: ReturnType, + instanceId = 'default' +): unknown[] { + const prefix = `iptvnator-server:unsent-writes:${instanceId}:`; + const key = [...local.values.keys()].find((name) => + name.startsWith(prefix) + ); + return key ? JSON.parse(local.values.get(key) ?? '[]') : []; +} + async function settle(): Promise { for (let i = 0; i < 10; i += 1) { await Promise.resolve(); @@ -470,13 +482,7 @@ describe('ServerUserDataStorage', () => { ); await first.storage.flush(); expect(first.storage.hasUnsentWrites()).toBe(true); - expect( - JSON.parse( - first.local.values.get( - 'iptvnator-server:unsent-writes:default' - ) ?? '[]' - ) - ).toMatchObject([ + expect(savedWrites(first.local)).toMatchObject([ { namespace: 'xtream-favorites', playlistId: 'p1', @@ -506,9 +512,7 @@ describe('ServerUserDataStorage', () => { {} ); expect(reloaded.hasUnsentWrites()).toBe(false); - expect( - first.local.values.has('iptvnator-server:unsent-writes:default') - ).toBe(false); + expect(savedWrites(first.local)).toEqual([]); }); it('saves a write still travelling when the page hides', async () => { @@ -521,12 +525,7 @@ describe('ServerUserDataStorage', () => { await settle(); storage.flushWithKeepalive(); - expect( - JSON.parse( - local.values.get('iptvnator-server:unsent-writes:default') ?? - '[]' - ) - ).toMatchObject([ + expect(savedWrites(local)).toMatchObject([ { namespace: 'xtream-favorites', playlistId: 'p1', value: ['x'] }, ]); }); @@ -536,7 +535,11 @@ describe('ServerUserDataStorage', () => { local.setItem('iptvnator-server:cache-owner', 'old-db'); local.setItem('iptvnator-server:xtream-favorites', '{"p1":["a"]}'); local.setItem( - 'iptvnator-server:unsent-writes:old-db', + 'iptvnator-server:unconfirmed:old-db', + JSON.stringify({ tab: { at: 0, keys: ['xtream-favorites|p1'] } }) + ); + local.setItem( + 'iptvnator-server:unsent-writes:old-db:tab', JSON.stringify([ { namespace: 'xtream-favorites', @@ -557,7 +560,8 @@ describe('ServerUserDataStorage', () => { expect(local.values.has('iptvnator-server:xtream-favorites')).toBe( false ); - expect(local.values.has('iptvnator-server:unsent-writes:old-db')).toBe( + expect(savedWrites(local, 'old-db')).toEqual([]); + expect(local.values.has('iptvnator-server:unconfirmed:old-db')).toBe( false ); expect(local.values.get('iptvnator-server:cache-owner')).toBe('new-db'); @@ -582,12 +586,7 @@ describe('ServerUserDataStorage', () => { await settle(); expect(storage.hasUnsentWrites()).toBe(true); - expect( - JSON.parse( - local.values.get('iptvnator-server:unsent-writes:default') ?? - '[]' - ) - ).toMatchObject([ + expect(savedWrites(local)).toMatchObject([ { 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 3ff49b553..86d8860c6 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 @@ -6,17 +6,28 @@ import { } from '@iptvnator/shared/interfaces'; import type { UserDataStorage } from '../user-data-storage'; import type { ServerApiClient } from './server-api-client'; -import { parseMap, splitEditKey } from './server-user-data-entries'; -import { type DeferredEdit, refreshedNamespace } from './server-user-data-edit'; +import { + parseMap, + splitEditKey, + UNKNOWN_TAG, +} from './server-user-data-entries'; +import { + type DeferredEdit, + refreshedNamespace, + replayWrite, +} from './server-user-data-edit'; import { UserDataWriteQueue } from './server-user-data-queue'; import { claimCache, markUnconfirmed, + orphanedPages, othersUnconfirmed, readDeferredEdits, readUnsentWrites, + removePage, saveDeferredEdits, saveUnsentWrites, + swapPage, } from './server-unsent-writes'; /** Server-mode cache keys; the legacy unprefixed keys are never touched. */ @@ -45,7 +56,7 @@ export class ServerUserDataStorage implements UserDataStorage { */ private readonly deferred = new Map(); private hydrated = false; - /** This tab, in the shared registry of unconfirmed entries. */ + /** This page load: names its saved writes and its registry record. */ private readonly tabId = Math.random().toString(36).slice(2); /** Cache values that did not fit into browser storage. */ private readonly overflow = new Map(); @@ -59,11 +70,10 @@ export class ServerUserDataStorage implements UserDataStorage { /** Scopes the cache and saved writes to one server database. */ private readonly instanceId = 'default', /** - * Where this tab keeps what the server has not confirmed: its own - * (`sessionStorage` in the app), so tabs never clear each other's, - * and a reload of the tab finds it again. + * The tab's own storage (`sessionStorage` in the app): remembers + * its page load, so a reload takes over the writes it saved. */ - private readonly tabStore: UserDataStorage = local + tabStore: UserDataStorage = local ) { this.writes = new UserDataWriteQueue(api, { changed: () => this.saveUnsent(), @@ -72,22 +82,81 @@ export class ServerUserDataStorage implements UserDataStorage { 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)) { + const previous = swapPage(tabStore, instanceId, this.tabId); + if (previous) { + this.takeOver(previous); + } + } + + /** + * A reload of this tab: the writes its last page load could not confirm + * go out with the first flush, and the first read keeps them. + */ + private takeOver(page: string): void { + for (const write of readUnsentWrites( + this.local, + this.instanceId, + page + )) { this.writes.queue( write.namespace, write.playlistId, write.value, write.before, - write.tag, - write.prior + { tag: write.tag, prior: write.prior } ); } - this.unsentSaved = this.writes.hasUnsent(); - for (const edit of readDeferredEdits(tabStore, instanceId)) { + for (const edit of readDeferredEdits( + this.local, + this.instanceId, + page + )) { this.deferred.set(`${edit.namespace}|${edit.playlistId}`, edit); } + // Saved under this page load before the old one is forgotten. + this.saveDeferred(); + this.saveUnsent(); + this.publishUnconfirmed(); + removePage(this.local, this.instanceId, page); + } + + /** + * Saved writes and early edits of closed tabs: replayed as this tab's + * own edits on what it shows now, then forgotten there. + */ + private adoptClosedTabs(): void { + for (const page of orphanedPages( + this.local, + this.instanceId, + this.tabId + )) { + const edits = [ + ...readDeferredEdits(this.local, this.instanceId, page).map( + (edit) => ({ ...edit, value: edit.after }) + ), + ...readUnsentWrites(this.local, this.instanceId, page), + ]; + for (const edit of edits) { + const before = this.baseline(edit.namespace); + const after = { ...before }; + const value = replayWrite(before[edit.playlistId], edit); + 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); + } + removePage(this.local, this.instanceId, page); + } + } + + /** Tells other tabs this one is alive and still sending its writes. */ + touch(): void { + if (this.hasUnsentWrites()) { + this.publishUnconfirmed(); + } } /** Whether a write is queued, being retried, or still travelling. */ @@ -177,9 +246,7 @@ export class ServerUserDataStorage implements UserDataStorage { this.deferred.delete(key); } } - saveDeferredEdits(this.tabStore, this.instanceId, [ - ...this.deferred.values(), - ]); + this.saveDeferred(); this.writes.drop(matches, playlistId); for (const namespace of SERVER_USER_STATE_NAMESPACES) { const map = this.baseline(namespace); @@ -232,7 +299,7 @@ export class ServerUserDataStorage implements UserDataStorage { // next write must conflict and be replayed on the server's. const fresh = { ...tags[namespace] }; for (const playlistId of keepCached) { - delete fresh[playlistId]; + fresh[playlistId] = UNKNOWN_TAG; } this.writes.setTags(namespace, fresh); } @@ -250,9 +317,10 @@ export class ServerUserDataStorage implements UserDataStorage { entries[edit.namespace]?.[edit.playlistId] ); } - saveDeferredEdits(this.tabStore, this.instanceId, []); + this.saveDeferred(); this.writes.scheduleFlush(); } + this.adoptClosedTabs(); } private readCache(namespace: ServerUserStateNamespace): string | null { @@ -314,13 +382,18 @@ export class ServerUserDataStorage implements UserDataStorage { next: Record, previous: Record ): void { + // The cache shows other tabs' unsent changes too. + const foreign = this.hydrated + ? othersUnconfirmed(this.local, this.instanceId, this.tabId) + : new Set(); const change = (playlistId: string, value: unknown) => { if (this.hydrated) { this.writes.queue( namespace, playlistId, value, - previous[playlistId] + previous[playlistId], + { foreign: foreign.has(`${namespace}|${playlistId}`) } ); return; } @@ -333,9 +406,7 @@ export class ServerUserDataStorage implements UserDataStorage { before: earlier ? earlier.before : previous[playlistId], after: value, }); - saveDeferredEdits(this.tabStore, this.instanceId, [ - ...this.deferred.values(), - ]); + this.saveDeferred(); this.publishUnconfirmed(); }; for (const [playlistId, value] of Object.entries(next)) { @@ -360,11 +431,17 @@ export class ServerUserDataStorage implements UserDataStorage { if (writes.length === 0 && !this.unsentSaved) { return; } - saveUnsentWrites(this.tabStore, this.instanceId, writes); + saveUnsentWrites(this.local, this.instanceId, this.tabId, writes); this.unsentSaved = writes.length > 0; this.publishUnconfirmed(); } + private saveDeferred(): void { + saveDeferredEdits(this.local, this.instanceId, this.tabId, [ + ...this.deferred.values(), + ]); + } + private publishUnconfirmed(): void { markUnconfirmed(this.local, this.instanceId, this.tabId, [ ...this.writes.unconfirmedKeys(), 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 2ce301391..cc61185b8 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 @@ -452,4 +452,86 @@ describe('ServerUserDataStorage with other devices', () => { [...(server.state['xtream-favorites']['p1'] as string[])].sort() ).toEqual(['S', 'X', 'Y', 'Z']); }); + + it('keeps a removal made after another tab built on the unsent item', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + 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(); + + tabA.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + // Tab B sees X in the shared cache and adds Y next to it. + tabB.setItem( + 'xtream-favorites', + JSON.stringify({ p1: [...favorites(tabB).p1, 'Y'] }) + ); + // Tab A takes X back before anything was sent. + tabA.setItem( + 'xtream-favorites', + JSON.stringify({ + p1: favorites(tabA).p1.filter((id: string) => id !== 'X'), + }) + ); + for (let round = 0; round < 4; round += 1) { + await tabB.flush(); + await tabA.flush(); + } + + expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'Y']); + }); + + it('sends the saved writes of a tab that was closed offline', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const shared = memoryStorage(); + const offline = { + ...server.api, + sendJson: jest + .fn() + .mockRejectedValue(new ServerApiError(0, 'network', 'offline')), + }; + const closed = new ServerUserDataStorage( + offline as never, + shared, + jest.fn(), + 'default', + memoryStorage() + ); + await closed.hydrate(); + closed.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + await closed.flush(); + // The tab is gone: its record stops being renewed. + const registry = 'iptvnator-server:unconfirmed:default'; + const records = JSON.parse(shared.values.get(registry) ?? '{}'); + for (const record of Object.values(records)) { + (record as { at: number }).at = 0; + } + shared.setItem(registry, JSON.stringify(records)); + + const next = new ServerUserDataStorage( + server.api as never, + shared, + jest.fn(), + 'default', + memoryStorage() + ); + await next.hydrate(); + await next.flush(); + + expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'X']); + expect( + [...shared.values.keys()].filter((key) => + key.startsWith('iptvnator-server:unsent-writes:') + ) + ).toEqual([]); + }); });