diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index 43e13cee6..a51b01c55 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -210,8 +210,8 @@ settings and guards read them in every build. per page load (`iptvnator-server:unsent-writes::`). The tab remembers its page load in `sessionStorage`, so a reload takes its writes over and sends them first; the writes of a tab closed before it - could send them are replayed by the next re-read in another tab once - that tab's record has gone stale. Tabs never clear each other's. + could send them are queued, ahead of its own, by the next re-read in + another tab once that tab's record has gone stale. Tabs never clear each other's. Cache and saved writes belong to one server database (`iptvnator-server:cache-owner`); another one behind the same address starts clean. Also, @@ -222,8 +222,9 @@ settings and guards read them in every build. replayed on the server's current value, so another device's change to the same playlist's entry is kept. A queued write keeps its edits as steps (before → after, also in the saved writes): an edit continuing the last - one joins it, one made behind a travelling write or on another tab's - change is kept apart, and a conflict replays them in order. So a lost + one joins it; one made behind a travelling write, after a write whose + answer was lost, or on another tab's change is kept apart; and a + conflict replays them in order. So a lost write is not dropped, and adding and then removing an item still removes it when the addition landed. When two tabs edit one entry before either write lands, a removal of the other tab's unsent item holds if that tab's 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 66008853d..de9b323fd 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 @@ -27,10 +27,19 @@ export interface PendingEntry { export interface EditStep { readonly before: unknown; readonly after: unknown; - /** The travelling write that carries it, until that one is answered. */ + /** + * The travelling write that carries it, until that one is answered; + * `0` once a write that carried it was lost or saved for a later page: + * it may have landed, so a later edit never joins it. + */ readonly sent?: number; } +/** Steps whose write may or may not have landed. */ +export function uncertain(steps: readonly EditStep[]): EditStep[] { + return steps.map((step) => ({ ...step, sent: 0 })); +} + /** * `steps` plus the edit `before` → `after`: joined with the last step when * it continues it, kept apart otherwise. 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 21120fd4c..80aa4b961 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 @@ -11,9 +11,11 @@ import { type EditStep, type PendingEntry, preconditionHeaders, + uncertain, UNKNOWN_TAG, } from './server-user-data-entries'; import { replaySteps } from './server-user-data-edit'; +import { UserStateTags } from './server-user-data-tags'; import { newestWrites } from './server-unsent-writes'; const FLUSH_DELAY_MS = 500; @@ -44,10 +46,7 @@ export interface UserDataQueueHooks { */ export class UserDataWriteQueue { /** Server version tag per entry; absent until the first read. */ - private readonly tags = new Map< - ServerUserStateNamespace, - Record - >(); + private readonly tags = new UserStateTags(); private readonly pending = new Map(); /** Writes still travelling (normal or page-hide), by entry key. */ private readonly inFlight = new Map(); @@ -114,7 +113,7 @@ export class UserDataWriteQueue { namespace: ServerUserStateNamespace, tags: Readonly> ): void { - this.tags.set(namespace, { ...tags }); + this.tags.set(namespace, tags); } /** @@ -168,8 +167,10 @@ export class UserDataWriteQueue { tag: base.tag !== undefined ? base.tag - : this.tagFor(namespace, playlistId), - steps: base.steps ?? [{ before, after: value }], + : this.tags.of(namespace, playlistId), + steps: base.steps + ? uncertain(base.steps) + : [{ before, after: value }], }; } if (base.foreign) { @@ -179,6 +180,28 @@ export class UserDataWriteQueue { this.edits.set(key, edit); } + /** + * A closed tab's saved edits, ahead of this tab's own (which may have + * been made on them): sent with them, replayed first on a conflict. + */ + adopt( + namespace: ServerUserStateNamespace, + playlistId: string, + value: unknown, + steps: readonly EditStep[], + foreign: boolean + ): void { + this.queue(namespace, playlistId, value, value, { foreign }); + const key = `${namespace}|${playlistId}`; + const entry = this.pending.get(key); + if (entry) { + this.pending.set(key, { + ...entry, + steps: [...uncertain(steps), ...entry.steps], + }); + } + } + scheduleFlush(): void { if (this.pending.size === 0) { return; @@ -271,41 +294,10 @@ export class UserDataWriteQueue { } } } - for (const [namespace, tags] of this.tags) { - if (playlistId === undefined) { - this.tags.set(namespace, {}); - } else { - delete tags[playlistId]; - } - } + this.tags.forget(playlistId); this.hooks.changed(); } - /** `undefined` until the server's tags are known (or not offered). */ - private tagFor( - namespace: ServerUserStateNamespace, - playlistId: string - ): string | null | undefined { - const tags = this.tags.get(namespace); - return tags ? (tags[playlistId] ?? null) : undefined; - } - - private setTag( - namespace: ServerUserStateNamespace, - playlistId: string, - tag: string | null - ): void { - const tags = this.tags.get(namespace); - if (!tags) { - return; - } - if (tag === null) { - delete tags[playlistId]; - } else { - tags[playlistId] = tag; - } - } - private takePending(): PendingEntry[] { if (this.flushTimer !== null) { clearTimeout(this.flushTimer); @@ -328,7 +320,7 @@ export class UserDataWriteQueue { private requeue(entry: PendingEntry): void { const key = entryKey(entry); if (this.edits.get(key) === entry.edit && !this.pending.has(key)) { - this.pending.set(key, entry); + this.pending.set(key, { ...entry, steps: uncertain(entry.steps) }); } } @@ -424,7 +416,7 @@ export class UserDataWriteQueue { /** The server took `entry`: the edit queued behind it builds on it. */ private confirmed(entry: PendingEntry, tag: string | null): void { - this.setTag(entry.namespace, entry.playlistId, tag); + this.tags.update(entry.namespace, entry.playlistId, tag); const key = entryKey(entry); const next = this.pending.get(key); if (next?.steps.some((step) => step.sent === entry.edit)) { @@ -446,7 +438,7 @@ export class UserDataWriteQueue { entry: PendingEntry, current: ServerUserStateCurrent ): void { - this.setTag(entry.namespace, entry.playlistId, current.tag); + this.tags.update(entry.namespace, entry.playlistId, current.tag); const key = entryKey(entry); const latest = this.pending.get(key) ?? 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 9b9376bd0..958993169 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 @@ -122,8 +122,8 @@ export class ServerUserDataStorage implements UserDataStorage { } /** - * Saved writes and early edits of closed tabs: replayed as this tab's - * own edits on what it shows now, then forgotten there. + * Saved writes and early edits of closed tabs: shown on what this tab + * shows now and queued with its writes, then forgotten there. */ private adoptClosedTabs(): void { for (const page of orphanedPages( @@ -135,18 +135,28 @@ export class ServerUserDataStorage implements UserDataStorage { ...readDeferredEdits(this.local, this.instanceId, page), ...readUnsentWrites(this.local, this.instanceId, page), ]; - for (const edit of edits) { - const before = this.baseline(edit.namespace); - const after = { ...before }; - const value = replaySteps(before[edit.playlistId], edit.steps); - if (value === undefined) { - delete after[edit.playlistId]; - } else { - after[edit.playlistId] = value; - } - this.writeCache(edit.namespace, JSON.stringify(after)); - this.queueChanges(edit.namespace, after, before); + const foreign = othersUnconfirmed( + this.local, + this.instanceId, + this.tabId + ); + for (const { namespace, playlistId, steps } of edits) { + const value = replaySteps( + this.baseline(namespace)[playlistId], + steps + ); + this.showEntry(namespace, playlistId, value); + // Queued even when the cache shows it already. + this.writes.adopt( + namespace, + playlistId, + value, + steps, + foreign.has(`${namespace}|${playlistId}`) + ); } + // Saved here before the closed tab's copy is forgotten. + this.writes.scheduleFlush(); removePage(this.local, this.instanceId, page); } } 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 d598f233e..409ed1822 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 @@ -1,4 +1,5 @@ import { ServerApiError } from './server-api-client'; +import { UNCONFIRMED_TTL_MS } from './server-unsent-writes'; import { ServerUserDataStorage } from './server-user-data-storage'; function memoryStorage() { @@ -593,4 +594,88 @@ describe('ServerUserDataStorage with other devices', () => { expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'X']); }); + + it('keeps a removal made after a write whose answer was lost', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const phone = new ServerUserDataStorage( + server.api as never, + memoryStorage(), + jest.fn() + ); + await phone.hydrate(); + const send = server.api.sendJson.getMockImplementation(); + if (!send) { + throw new Error('The fake server answers sendJson'); + } + // The server saves X, but the answer never arrives. + server.api.sendJson.mockImplementationOnce( + async (...args: Parameters) => { + await send(...args); + throw new ServerApiError(0, 'network', 'offline'); + } + ); + phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + await phone.flush(); + expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'X']); + + phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S'] })); + await phone.flush(); + await phone.flush(); + + expect(server.state['xtream-favorites']['p1']).toEqual(['S']); + }); + + it("sends a closed tab's favorite this tab's cache already shows", async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const shared = memoryStorage(); + const send = server.api.sendJson.getMockImplementation(); + if (!send) { + throw new Error('The fake server answers sendJson'); + } + const offline = () => + Promise.reject(new ServerApiError(0, 'network', 'offline')); + let online = false; + const tabApi = { + ...server.api, + sendJson: jest.fn((...args: Parameters) => + online ? send(...args) : offline() + ), + }; + const clock = jest.spyOn(Date, 'now').mockReturnValue(1_000); + const closed = new ServerUserDataStorage( + { ...server.api, sendJson: jest.fn(offline) } as never, + shared, + jest.fn(), + 'default', + memoryStorage() + ); + await closed.hydrate(); + closed.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + await closed.flush(); + const tab = new ServerUserDataStorage( + tabApi as never, + shared, + jest.fn(), + 'default', + memoryStorage() + ); + await tab.hydrate(); + tab.setItem( + 'xtream-favorites', + JSON.stringify({ p1: [...favorites(tab).p1, 'Y'] }) + ); + await tab.flush(); + + // Later the closed tab's record is stale; this tab is still offline. + clock.mockReturnValue(1_000 + UNCONFIRMED_TTL_MS + 1); + await tab.hydrate(); + online = true; + await tab.flush(); + await tab.flush(); + clock.mockRestore(); + + expect( + [...(server.state['xtream-favorites']['p1'] as string[])].sort() + ).toEqual(['S', 'X', 'Y']); + }); }); diff --git a/libs/services/src/lib/server-storage/server-user-data-tags.ts b/libs/services/src/lib/server-storage/server-user-data-tags.ts new file mode 100644 index 000000000..7172eea77 --- /dev/null +++ b/libs/services/src/lib/server-storage/server-user-data-tags.ts @@ -0,0 +1,54 @@ +import type { ServerUserStateNamespace } from '@iptvnator/shared/interfaces'; + +/** The server version tag of each user-state entry, by namespace. */ +export class UserStateTags { + /** Absent for a namespace until the first read. */ + private readonly tags = new Map< + ServerUserStateNamespace, + Record + >(); + + set( + namespace: ServerUserStateNamespace, + tags: Readonly> + ): void { + this.tags.set(namespace, { ...tags }); + } + + /** `undefined` until the server's tags are known (or not offered). */ + of( + namespace: ServerUserStateNamespace, + playlistId: string + ): string | null | undefined { + const tags = this.tags.get(namespace); + return tags ? (tags[playlistId] ?? null) : undefined; + } + + /** `null`: the entry is gone. */ + update( + namespace: ServerUserStateNamespace, + playlistId: string, + tag: string | null + ): void { + const tags = this.tags.get(namespace); + if (!tags) { + return; + } + if (tag === null) { + delete tags[playlistId]; + } else { + tags[playlistId] = tag; + } + } + + /** After a delete: one playlist's tags, without an id all of them. */ + forget(playlistId?: string): void { + for (const [namespace, tags] of this.tags) { + if (playlistId === undefined) { + this.tags.set(namespace, {}); + } else { + delete tags[playlistId]; + } + } + } +}