diff --git a/docs/architecture/pwa-self-hosted.md b/docs/architecture/pwa-self-hosted.md index b69eeb347..e89adb84e 100644 --- a/docs/architecture/pwa-self-hosted.md +++ b/docs/architecture/pwa-self-hosted.md @@ -182,7 +182,9 @@ instance under `iptvnator-server:instance`. When the probe gets no answer (network error, timeout, or a 5xx such as nginx's 502 while the backend restarts), a browser that remembers this backend still starts in server mode, offline, and reloads once the server answers; it never falls back to -an empty browser library. An answer without server storage, or a 404 from +an empty browser library. Until that reload it sends no writes +(`holdWrites`): the address may lead to another database by then, which +the reload's probe finds out. An answer without server storage, or a 404 from an older backend, forgets it (`server-storage-start.ts`). Everything else of the client lives behind the `@iptvnator/services/server-storage` sub-entry, so other builds never load it; `ServerStorageStatusService` and the token @@ -215,11 +217,16 @@ settings and guards read them in every build. as offline. Each write is sent on the tag of the entry it was made against (`UserDataWriteQueue`); after a `412` the edit (before → after) is replayed on the server's current value, so another device's change to the - same playlist's entry is kept. An entry a re-read keeps from the cache - gets no tag, so its next write conflicts and is replayed too. A page-hide - write behind a travelling one 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 + same playlist's entry is kept. An edit queued behind a travelling write + 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 + 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 diff --git a/libs/services/src/lib/server-storage/server-api-client.spec.ts b/libs/services/src/lib/server-storage/server-api-client.spec.ts index fd6fb6a52..1ff88268d 100644 --- a/libs/services/src/lib/server-storage/server-api-client.spec.ts +++ b/libs/services/src/lib/server-storage/server-api-client.spec.ts @@ -84,19 +84,56 @@ describe('ServerApiClient', () => { expect(status.reportOffline).not.toHaveBeenCalled(); }); - it('reports whether a keepalive write was taken', async () => { - const ok = jest.fn().mockResolvedValue(response(200, {})); + it('reports how a keepalive write ended', async () => { + const ok = jest.fn().mockResolvedValue(response(200, { tag: 't2' })); await expect( client(ok).sendKeepalive('PUT', '/user-state/a/b', { value: 1 }) - ).resolves.toBe(true); + ).resolves.toEqual({ status: 'saved', tag: 't2' }); expect(ok.mock.calls[0][1]).toMatchObject({ keepalive: true }); + const conflict = jest.fn().mockResolvedValue( + response(412, { + code: 'conflict', + message: 'changed', + current: { value: [2], tag: 't3' }, + }) + ); + await expect( + client(conflict).sendKeepalive('PUT', '/user-state/a/b', { + value: 1, + }) + ).resolves.toEqual({ + status: 'conflict', + current: { value: [2], tag: 't3' }, + }); + const refused = jest .fn() .mockRejectedValue(new TypeError('body too large')); await expect( client(refused).sendKeepalive('DELETE', '/user-state/a/b') - ).resolves.toBe(false); + ).resolves.toEqual({ status: 'failed' }); + }); + + it('holds writes, not reads, until the instance is checked', async () => { + const fetchFn = jest.fn().mockResolvedValue(response(200, {})); + const held = new ServerApiClient({ + baseUrl: '/api', + status, + fetchFn: fetchFn as unknown as typeof fetch, + holdWrites: true, + }); + + await expect( + held.sendJson('PUT', '/user-state/a/b', { value: 1 }) + ).rejects.toMatchObject({ status: 0, code: 'network' }); + await expect( + held.sendKeepalive('DELETE', '/user-state/a/b') + ).resolves.toEqual({ status: 'failed' }); + await held.getJson('/instance'); + + expect(fetchFn).toHaveBeenCalledTimes(1); + expect(fetchFn.mock.calls[0][0]).toBe('/api/v1/instance'); }); it('reports offline when a body is cut off after the headers', async () => { diff --git a/libs/services/src/lib/server-storage/server-api-client.ts b/libs/services/src/lib/server-storage/server-api-client.ts index c3cae2b28..56fdaad5d 100644 --- a/libs/services/src/lib/server-storage/server-api-client.ts +++ b/libs/services/src/lib/server-storage/server-api-client.ts @@ -1,6 +1,7 @@ import type { ServerStorageErrorBody, ServerUserStateCurrent, + ServerUserStateWriteResponse, } from '@iptvnator/shared/interfaces'; import { clearServerAccessToken, @@ -26,6 +27,12 @@ export class ServerApiError extends Error { } } +/** How a page-hide write ended, as far as the page could see. */ +export type KeepaliveOutcome = + | { readonly status: 'saved'; readonly tag?: string | null } + | { readonly status: 'conflict'; readonly current: ServerUserStateCurrent } + | { readonly status: 'failed' }; + export interface ServerApiClientOptions { readonly baseUrl: string; readonly status: Pick< @@ -36,6 +43,12 @@ export interface ServerApiClientOptions { readonly onWriteFailure?: (error: ServerApiError) => void; readonly fetchFn?: typeof fetch; readonly timeoutMs?: number; + /** + * Refuse writes as if offline: a start from a remembered instance that + * has not answered yet, whose address may lead to another database by + * the time it does. + */ + readonly holdWrites?: boolean; } /** `fetch` for `/v1`: token, timeout, error mapping, connection state. */ @@ -54,6 +67,9 @@ export class ServerApiClient { const isWrite = (init.method ?? 'GET') !== 'GET'; let response: Response; try { + if (isWrite && this.options.holdWrites) { + throw new Error('Writes wait for the instance to answer.'); + } response = await this.fetchFn()(this.url(path), { ...init, cache: 'no-store', @@ -150,15 +166,18 @@ export class ServerApiClient { } /** - * A write that survives the page. Resolves whether the server took it; - * browsers reject keepalive bodies above 64 KiB, which resolves false. + * A write that survives the page. Browsers reject keepalive bodies + * above 64 KiB, which ends as `failed`. */ async sendKeepalive( method: 'PUT' | 'DELETE', path: string, body?: unknown, headers: Record = {} - ): Promise { + ): Promise { + if (this.options.holdWrites) { + return { status: 'failed' }; + } try { const response = await this.fetchFn()(this.url(path), { method, @@ -172,9 +191,21 @@ export class ServerApiClient { }, ...(body === undefined ? {} : { body: JSON.stringify(body) }), }); - return response.ok; + if (response.ok) { + const body = (await response + .json() + .catch( + () => ({}) + )) as Partial; + return { status: 'saved', tag: body?.tag }; + } + const { current } = + response.status === 412 ? await readErrorBody(response) : {}; + return current + ? { status: 'conflict', current } + : { status: 'failed' }; } catch { - return false; + return { status: 'failed' }; } } 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 41fa1e5f4..46e6b3a38 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 @@ -29,6 +29,7 @@ describe('ServerStorageSession', () => { jest.useRealTimers(); fetchMock.mockReset(); localStorage.clear(); + sessionStorage.clear(); TestBed.resetTestingModule(); }); @@ -126,6 +127,37 @@ describe('ServerStorageSession', () => { expect(reload).toHaveBeenCalledTimes(1); }); + 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', + JSON.stringify([ + { + namespace: 'xtream-favorites', + playlistId: 'p1', + value: ['x'], + }, + ]) + ); + const { session } = setup('none', true); + jest.spyOn( + session as unknown as { reloadPage(): void }, + 'reloadPage' + ).mockImplementation(() => undefined); + + await session.start(); + session.userData.flushWithKeepalive(); + await session.userData.flush(); + + expect( + fetchMock.mock.calls.map(([url, init]) => [ + url, + init?.method ?? 'GET', + ]) + ).toEqual([['/api/v1/instance', 'GET']]); + expect(session.userData.hasUnsentWrites()).toBe(true); + }); + it('never reloads after a normal start', async () => { const { session } = setup('none'); const reload = jest 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 222a67485..14ef3d5ca 100644 --- a/libs/services/src/lib/server-storage/server-storage-session.ts +++ b/libs/services/src/lib/server-storage/server-storage-session.ts @@ -2,7 +2,10 @@ import { effect, inject, Injectable } from '@angular/core'; import { MatSnackBar } from '@angular/material/snack-bar'; import { Router } from '@angular/router'; import { TranslateService } from '@ngx-translate/core'; -import type { ServerStorageStatusResponse } from '@iptvnator/shared/interfaces'; +import type { + ServerInstanceInfo, + ServerStorageStatusResponse, +} from '@iptvnator/shared/interfaces'; import { PlaylistStorageBackend } from '../playlist-storage-backend'; import { clearServerAccessToken, @@ -39,6 +42,9 @@ export class ServerStorageSession { baseUrl: this.config.baseUrl, status: this.status, onWriteFailure: (error) => this.announceWriteFailure(error), + // A remembered instance may answer as another database: writes + // wait for the reload that checks which one it is. + holdWrites: this.config.startedOffline, }); readonly documentStore = new ServerPlaylistDocumentStore( this.api, @@ -83,14 +89,28 @@ export class ServerStorageSession { /** Re-reads synced user data; also the offline retry. */ async refresh(): Promise { + if (this.config.startedOffline) { + await this.reloadOnceReachable(); + return; + } try { await this.userData.hydrate(); } catch { // The API client already reported offline or unauthorized. + } + } + + /** + * Nothing was read at start: once the server answers, boot again, so + * its instance is checked before anything is sent. + */ + private async reloadOnceReachable(): Promise { + try { + await this.api.getJson('/instance'); + } catch { return; } - if (this.config.startedOffline && !this.reloading) { - // Nothing was read at start: boot again against the server. + if (!this.reloading) { this.reloading = true; this.reloadPage(); } 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 7c76d5d15..c5063fad9 100644 --- a/libs/services/src/lib/server-storage/server-unsent-writes.ts +++ b/libs/services/src/lib/server-storage/server-unsent-writes.ts @@ -65,6 +65,8 @@ export interface UnsentWrite { readonly before?: unknown; /** The server version it is based on (see `PendingEntry.tag`). */ readonly tag?: string | null; + /** A write sent before it, maybe saved (see `PendingEntry.prior`). */ + readonly prior?: { readonly before: unknown; readonly after: unknown }; } interface StoredWrite { @@ -74,6 +76,8 @@ interface StoredWrite { readonly remove?: true; readonly before?: unknown; readonly tag?: string | null; + /** JSON drops an undefined side: absent reads back as undefined. */ + readonly prior?: { readonly before?: unknown; readonly after?: unknown }; } export function readUnsentWrites( @@ -101,6 +105,13 @@ export function readUnsentWrites( value: write.remove ? undefined : write.value, before: write.before, tag: write.tag, + prior: + write.prior && typeof write.prior === 'object' + ? { + before: write.prior.before, + after: write.prior.after, + } + : undefined, }); } } @@ -118,12 +129,15 @@ export function saveUnsentWrites( return; } const stored: StoredWrite[] = writes.map( - ({ namespace, playlistId, value, before, tag }) => ({ + ({ namespace, playlistId, value, before, tag, prior }) => ({ namespace, playlistId, ...(value === undefined ? { remove: true } : { value }), ...(before === undefined ? {} : { before }), ...(tag === undefined ? {} : { tag }), + ...(prior === undefined + ? {} + : { prior: { before: prior.before, after: prior.after } }), }) ); local.setItem(writesKey(instanceId), JSON.stringify(stored)); 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 51decac5c..1cd75e875 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 @@ -15,6 +15,19 @@ export interface PendingEntry { * seen absent, `undefined` when unknown (an unconditional write). */ readonly tag?: string | null; + /** + * The edit of the write this one was queued behind, which may or may + * not have reached the server: a conflict replays it first. Kept apart + * from this edit, so adding and then removing an item is not one no-op. + */ + readonly prior?: PriorEdit; +} + +export interface PriorEdit { + readonly before: unknown; + readonly after: unknown; + /** The write it was sent with; `0` after a reload. */ + readonly edit: number; } /** `If-Match` / `If-None-Match` for a write based on `tag`. */ 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 6a889a5d1..30a0f7ed9 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 @@ -9,6 +9,7 @@ import { entryPath, type PendingEntry, preconditionHeaders, + type PriorEdit, } from './server-user-data-entries'; import { applyEdit } from './server-user-data-edit'; import { newestWrites } from './server-unsent-writes'; @@ -115,28 +116,49 @@ export class UserDataWriteQueue { } /** - * A newer edit of an entry that is still queued or travelling keeps - * that write's baseline and tag, so a conflict replays both edits. + * 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. */ queue( namespace: ServerUserStateNamespace, playlistId: string, value: unknown, before: unknown, - tag = this.tagFor(namespace, playlistId) + tag = this.tagFor(namespace, playlistId), + prior?: Omit ): void { const key = `${namespace}|${playlistId}`; - const earlier = this.pending.get(key) ?? this.inFlight.get(key); - const entry: PendingEntry = { - namespace, - playlistId, - value, - edit: ++this.editCount, - before: earlier ? earlier.before : before, - tag: earlier ? earlier.tag : tag, - }; + const queued = this.pending.get(key); + const travelling = this.inFlight.get(key); + const edit = ++this.editCount; + let entry: PendingEntry; + if (queued) { + entry = { ...queued, value, edit }; + } else if (travelling) { + entry = { + namespace, + playlistId, + value, + edit, + before, + tag: travelling.tag, + prior: { + before: travelling.prior + ? travelling.prior.before + : travelling.before, + after: travelling.value, + edit: travelling.edit, + }, + }; + } else { + entry = { namespace, playlistId, value, edit, before, tag }; + if (prior) { + entry = { ...entry, prior: { ...prior, edit: 0 } }; + } + } this.pending.set(key, entry); - this.edits.set(key, entry.edit); + this.edits.set(key, edit); } scheduleFlush(): void { @@ -322,7 +344,7 @@ export class UserDataWriteQueue { private async sendKeepalive(entry: PendingEntry): Promise { const path = entryPath(entry); const headers = preconditionHeaders(entry.tag); - const ok = await (entry.value === undefined + const outcome = await (entry.value === undefined ? this.api.sendKeepalive('DELETE', path, undefined, headers) : this.api.sendKeepalive( 'PUT', @@ -331,7 +353,14 @@ export class UserDataWriteQueue { headers )); this.retireInFlight(entry); - if (!ok) { + if (outcome.status === 'saved') { + // Without a tag a newer edit keeps this one as `prior`. + if (outcome.tag !== undefined) { + this.confirmed(entry, outcome.tag); + } + } else if (outcome.status === 'conflict') { + this.replayOnConflict(entry, outcome.current); + } else { this.requeue(entry); } this.hooks.changed(); @@ -375,20 +404,21 @@ export class UserDataWriteQueue { this.hooks.changed(); } - /** The server took `entry`: a newer queued edit now builds on it. */ + /** 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); const key = entryKey(entry); const next = this.pending.get(key); - if (next && next.before === entry.before && next.tag === entry.tag) { - this.pending.set(key, { ...next, before: entry.value, tag }); + if (next?.prior?.edit === entry.edit) { + this.pending.set(key, { ...next, prior: undefined, tag }); } } /** - * Another device changed the entry: apply this edit (before → after) to - * the server's current value, show the result and send it on that - * version, unless a newer edit of the entry will. + * Another device changed the entry and `entry` did not land: replay its + * edits on the server's current value (with the newer queued edit that + * carries it as `prior`, if any), show the result and send it on that + * version. */ private replayOnConflict( entry: PendingEntry, @@ -396,36 +426,35 @@ export class UserDataWriteQueue { ): void { this.setTag(entry.namespace, entry.playlistId, current.tag); const key = entryKey(entry); - const next = this.pending.get(key); - if (next) { - // The newer edit kept this one's baseline: replay both at once. - this.pending.set(key, { - ...next, - value: applyEdit(current.value, next.before, next.value), - before: current.value, - tag: current.tag, - }); - } else if (this.edits.get(key) === entry.edit) { - const merged = applyEdit(current.value, entry.before, entry.value); - if (JSON.stringify(merged) !== JSON.stringify(current.value)) { - this.pending.set(key, { - ...entry, - value: merged, - before: current.value, - tag: current.tag, - }); - } else { - this.hooks.show(entry.namespace, entry.playlistId, merged); - return; - } - } else { + const latest = + this.pending.get(key) ?? + (this.edits.get(key) === entry.edit ? entry : undefined); + if (!latest) { return; } - this.hooks.show( - entry.namespace, - entry.playlistId, - this.pending.get(key)?.value - ); + const merged = replay(current.value, latest); + this.hooks.show(entry.namespace, entry.playlistId, merged); + if ( + latest === entry && + JSON.stringify(merged) === JSON.stringify(current.value) + ) { + return; + } + this.pending.set(key, { + ...latest, + value: merged, + before: current.value, + tag: current.tag, + prior: undefined, + }); 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 9f6aa295c..0b7266e10 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 @@ -1,3 +1,4 @@ +import type { KeepaliveOutcome } from './server-api-client'; import { ServerUserDataStorage } from './server-user-data-storage'; function memoryStorage() { @@ -34,7 +35,7 @@ describe('ServerUserDataStorage', () => { }), sendJson: jest.fn().mockResolvedValue(undefined), request: jest.fn().mockResolvedValue(undefined), - sendKeepalive: jest.fn().mockResolvedValue(true), + sendKeepalive: jest.fn().mockResolvedValue({ status: 'saved' }), readJson: jest.fn().mockResolvedValue({}), }; const onWriteFailure = jest.fn(); @@ -269,7 +270,7 @@ describe('ServerUserDataStorage', () => { it('queues a page-hide write again when the server did not take it', async () => { const { api, storage } = setup(); await storage.hydrate(); - api.sendKeepalive.mockResolvedValueOnce(false); + api.sendKeepalive.mockResolvedValueOnce({ status: 'failed' }); storage.setItem('xtream-favorites', JSON.stringify({ p1: ['big'] })); storage.flushWithKeepalive(); @@ -310,9 +311,9 @@ describe('ServerUserDataStorage', () => { it('lets a newer write wait for a page-hide write still travelling', async () => { const { api, storage } = setup(); await storage.hydrate(); - let acknowledge: (ok: boolean) => void = () => undefined; + let acknowledge: (outcome: KeepaliveOutcome) => void = () => undefined; api.sendKeepalive.mockReturnValueOnce( - new Promise((resolve) => (acknowledge = resolve)) + new Promise((resolve) => (acknowledge = resolve)) ); storage.setItem('xtream-favorites', JSON.stringify({ p1: ['old'] })); @@ -323,7 +324,7 @@ describe('ServerUserDataStorage', () => { await settle(); expect(api.sendJson).not.toHaveBeenCalled(); - acknowledge(true); + acknowledge({ status: 'saved' }); await newer; expect(api.sendJson).toHaveBeenCalledWith( 'PUT', @@ -397,9 +398,9 @@ describe('ServerUserDataStorage', () => { it('never retries an older page-hide write over a newer one', async () => { const { api, storage } = setup(); await storage.hydrate(); - const acks: ((ok: boolean) => void)[] = []; + const acks: ((outcome: KeepaliveOutcome) => void)[] = []; api.sendKeepalive.mockImplementation( - () => new Promise((resolve) => acks.push(resolve)) + () => new Promise((resolve) => acks.push(resolve)) ); for (const value of ['A', 'B', 'C']) { @@ -410,7 +411,9 @@ describe('ServerUserDataStorage', () => { storage.flushWithKeepalive(); await settle(); // Each page-hide write waits for the previous one. - acks[acks.length - 1](value !== 'B'); + acks[acks.length - 1]({ + status: value === 'B' ? 'failed' : 'saved', + }); await settle(); } @@ -651,7 +654,7 @@ describe('ServerUserDataStorage', () => { }), sendJson: jest.fn().mockResolvedValue(undefined), request: jest.fn().mockResolvedValue(undefined), - sendKeepalive: jest.fn().mockResolvedValue(true), + sendKeepalive: jest.fn().mockResolvedValue({ status: 'saved' }), readJson: jest.fn().mockResolvedValue({}), }; const tab = () => 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 9d2a1fc5f..3ff49b553 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 @@ -80,7 +80,8 @@ export class ServerUserDataStorage implements UserDataStorage { write.playlistId, write.value, write.before, - write.tag + write.tag, + write.prior ); } this.unsentSaved = this.writes.hasUnsent(); 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 514498173..2ce301391 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 @@ -84,10 +84,16 @@ function fakeServer(initial: Record> = {}) { headers: Record = {} ) => { try { - write(path, headers, body?.value); - return true; - } catch { - return false; + const response = write(path, headers, body?.value); + const { tag } = await response.json(); + return { status: 'saved' as const, tag }; + } catch (error) { + return error instanceof ServerApiError && error.current + ? { + status: 'conflict' as const, + current: error.current, + } + : { status: 'failed' as const }; } } ), @@ -338,4 +344,112 @@ describe('ServerUserDataStorage with other devices', () => { expect(server.state['xtream-favorites']['p1']).toEqual(['S']); expect(phone.hasUnsentWrites()).toBe(false); }); + + it('removes an item again after a page-hide write added it', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const phone = new ServerUserDataStorage( + server.api as never, + memoryStorage(), + jest.fn() + ); + await phone.hydrate(); + + let answer: () => void = () => undefined; + const keepalive = server.api.sendKeepalive.getMockImplementation(); + if (!keepalive) { + throw new Error('The fake server answers sendKeepalive'); + } + server.api.sendKeepalive.mockImplementationOnce( + (...args: Parameters) => + new Promise( + (resolve) => (answer = () => resolve(keepalive(...args))) + ) + ); + phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + phone.flushWithKeepalive(); + // Back on the page, X is removed while the page-hide write travels. + phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S'] })); + answer(); + await phone.flush(); + await phone.flush(); + + expect(server.state['xtream-favorites']['p1']).toEqual(['S']); + expect(phone.hasUnsentWrites()).toBe(false); + }); + + it('recovers a removal queued behind a write whose answer was lost', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const local = memoryStorage(); + const tabStore = memoryStorage(); + const open = () => + new ServerUserDataStorage( + server.api as never, + local, + jest.fn(), + 'default', + tabStore + ); + const before = open(); + await before.hydrate(); + + const send = server.api.sendJson.getMockImplementation(); + if (!send) { + throw new Error('The fake server answers sendJson'); + } + // The server saves X, but the page reloads before the answer. + server.api.sendJson.mockImplementationOnce( + (...args: Parameters) => { + void send(...args); + return new Promise(() => undefined); + } + ); + before.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + void before.flush(); + await new Promise((resolve) => setTimeout(resolve, 0)); + before.setItem('xtream-favorites', JSON.stringify({ p1: ['S'] })); + expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'X']); + + const reloaded = open(); + await reloaded.hydrate(); + await reloaded.flush(); + + expect(server.state['xtream-favorites']['p1']).toEqual(['S']); + expect(reloaded.hasUnsentWrites()).toBe(false); + }); + + it('replays a failed write it was queued behind on a conflict', async () => { + const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } }); + const phone = new ServerUserDataStorage( + server.api as never, + memoryStorage(), + jest.fn() + ); + await phone.hydrate(); + + let fail: () => void = () => undefined; + server.api.sendJson.mockImplementationOnce( + () => + new Promise((_, reject) => { + fail = () => + reject(new ServerApiError(0, 'network', 'offline')); + }) + ); + phone.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] })); + const first = phone.flush(); + await new Promise((resolve) => setTimeout(resolve, 0)); + phone.setItem( + 'xtream-favorites', + JSON.stringify({ p1: ['S', 'X', 'Y'] }) + ); + // X never reached the server; another device saves Z meanwhile. + server.state['xtream-favorites']['p1'] = ['S', 'Z']; + fail(); + await first; + await phone.flush(); + await phone.flush(); + + expect( + [...(server.state['xtream-favorites']['p1'] as string[])].sort() + ).toEqual(['S', 'X', 'Y', 'Z']); + }); });