diff --git a/CHANGELOG.md b/CHANGELOG.md index 4811236e..b8b7defc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -116,6 +116,28 @@ pending until the returned function is called, so a test can model a native that is slow to answer. Later enqueues wait behind it, as they do on a device. +Added `useRequests(filter?)` and `useRequestsWithProgress(filter?)`, React +hooks that return the queue rows for display, and +`definition.useRequests()` and `definition.useRequestsWithProgress()`, which +return one definition's rows with typed `vars`. The client keeps one copy of +the rows in memory, read from native when the first hook mounts and updated +by each `state` event. It also removes a row when native forgets it with no +event: after the ack of a `completed` or `cancelled` outcome, and after +`cancel()` on a settled entry. A `state` event that arrives after the copy +removed its row is checked against native, so it does not add a finished row +back. The filter takes `includeCancelled`; cancelled rows are left out by +default. `addRequestsListener()` and `getCachedRequests()` give the same rows +outside React. Completed rows are included until their outcome is +acknowledged. An app that keeps its own copy of `getRequests()` for display, +and wraps `cancel()` to update it, can remove that code. Delivery, acks and +cancel are unchanged. `define()` now returns the two hooks with `key` and +`mutate`. The `Defined` and `UploadClient` types gained the new functions as +required members, so a test double built by hand with one of those types no +longer compiles until it adds them. + +`createFakeNative().settle()` now resolves one promise turn later, after the +client has handled the ack, so the row hooks are up to date when it resolves. + ## 11.1.0 Added `takeDiagnostics()`. On Android it returns the module's recent diff --git a/README.md b/README.md index 6e84bf0d..54cca23a 100644 --- a/README.md +++ b/README.md @@ -339,7 +339,7 @@ carries the last one. `createUploadClient()` returns a client. The default export is one client; an app needs one. -### `define(definition): { key, mutate }` +### `define(definition): { key, mutate, useRequests, useRequestsWithProgress }` ```ts type Definition = @@ -484,6 +484,89 @@ uploads .forEach((row) => uploads.cancel(row.id)); ``` +### `useRequests(filter?)` and `useRequestsWithProgress(filter?)` +React hooks that return the queue rows, for display. The client keeps one +copy of the rows in memory. It reads native when the first hook mounts, and +each `state` event updates the row it carries. Every component shares that +copy, so the app does not keep its own. + +```tsx +// Take the hooks out of the client and the definition at module level. The +// hooks lint rule checks `X.useY()` only when `X` starts with a capital +// letter, so `uploads.useRequests()` inside a condition is not flagged. +// `useRequests()` called as a plain name is checked. +const { useRequests } = uploads; +const { useRequests: useCaptureRows } = captureUpload; + +const CaptureBadge = ({ captureId }: { captureId: string }) => { + // Typed vars: the rows of this definition only. + const rows = useCaptureRows(); + const pending = rows.filter( + (row) => row.vars.captureId === captureId && row.state !== 'completed', + ); + return pending.length > 0 ? : null; +}; +``` + +- `filter` is `{ key?, id?, includeCancelled? }`. `key` and `id` match as in + `getRequests()`. Cancelled rows are left out unless `includeCancelled` is + `true`. A cancelled row has no work left, and native forgets it at its ack. +- Completed rows are included. A completed row stays until its outcome is + acknowledged. When its handler rejects or its key has no definition, the + outcome is not acknowledged, so the row stays for the whole session. Do not + treat a completed row as work still to do. +- `useRequests()` re-renders when a row appears, changes state, or goes. A + row's bytes are those of its last `state` event. +- `useRequestsWithProgress()` also re-renders on each `progress` event of a + row it returns, with that event's bytes. Use it only where bytes are shown. +- The result is the same array while the rows it holds are the same, even + when the filter is a new object each render. A component filtered by `id` + re-renders only when that row changes. The array and its rows are shared, + so do not change them. +- `definition.useRequests(filter?)` and `definition.useRequestsWithProgress(filter?)` + return the rows of that definition's key, with `vars` typed as its vars. + The type is not checked at run time: `vars` is what native stored, so a + `Date` reads back as a string, and a row enqueued under an older vars type + keeps its old shape. + +Native emits no event when it forgets a row, so the library removes the row +itself when its own call made native forget it: when the ack of a +`completed` or `cancelled` outcome resolves, and when `cancel(id)` resolves +on a settled entry. An `error` row stays until `cancel()` or a same-id +`mutate()`, as in native. A same-id `mutate()` emits a `state` event, so it +needs nothing extra. iOS can settle a live row during an ack with no event; +after any ack whose row the copy does not show in the outcome's state, the +library reads that row from native. It does the same after `cancel(id)` on a +row the copy shows live or `cancelled`. Events and call results can reach JS +in either order, so a `state` event can arrive after the library removed its +row (Android emits a run's outcome before its last `state` event). The +library keeps the last 256 ids it removed this way, and checks a `state` +event for one of them against native instead of adding the row back. Past +256 ids the result is the same: a settled `state` event for a row the copy +does not hold is also checked, at the cost of one native read. A same-id +`mutate()` of a kept id costs one native read. + +Limits. While no hook or listener is mounted, the copy is not updated. Each +time the first hook or listener subscribes, the library checks the copy +against native, so a mount can show old rows for one render. When the +native read fails, the library keeps the rows it has and sends a warning to +`onWarning`. Any other change that native makes with no event is missed +until the next check. + +### `addRequestsListener(listener)` and `getCachedRequests(filter?)` +The same rows outside React, for code such as a store or a logger. +`addRequestsListener` returns `{ remove }` and calls `listener` with +`getCachedRequests()` each time those rows change. There is no call when it +registers, so read `getCachedRequests()` after it returns. It keeps the copy +up to date as a mounted hook does. + +`getCachedRequests(filter?)` returns the rows the hooks see. While nothing +listens, it reads native on each call and returns a new array. While a hook +or listener is mounted and the filter has no `key` or `id`, it returns the +same array until a row changes. A result filtered by `key` or `id` is a new +array on each call, so do not use it as a `useSyncExternalStore` snapshot; +use the hooks. + ### `takeDiagnostics(): Promise` The library's persisted diagnostic records, oldest first. Each call returns the records since the last call and clears them, so send them somewhere @@ -707,6 +790,14 @@ The fake does not model the same-id rules (resume, replace, `E_RUNNING`, re-emit) or the header merge of `updateHeaders()`; script those with `failNext()` and `seedUnacknowledged()`. +The fake drives the row hooks as native does. `settle()` resolves after the +client has handled the ack, so a completed row is already gone from +`useRequests()`. `seedRows()` and `reset()` change rows with no event, as +leftovers from a dead session would. Call them before the first hook mounts +or listener subscribes, or unmount and mount again: the next mount checks the +rows against the fake. The hooks need a React renderer, for example +`react-test-renderer`, and its updates wrapped in `act()`. + `native.entries` holds every enqueue as `mutate()` sent it, with `vars` and `data` parsed back from JSON. `failNext()` makes the next native call reject with a code such as `E_STORAGE`. `holdNext('enqueue')` keeps the next enqueue diff --git a/example/RNBGUExample/__tests__/useRequests.test.tsx b/example/RNBGUExample/__tests__/useRequests.test.tsx new file mode 100644 index 00000000..b7690c46 --- /dev/null +++ b/example/RNBGUExample/__tests__/useRequests.test.tsx @@ -0,0 +1,247 @@ +/** + * @format + */ + +import 'react-native'; +import React from 'react'; + +// Note: import explicitly to use the types shipped with jest. +import {beforeEach, expect, it, jest} from '@jest/globals'; + +// Note: test renderer must be required after react-native. +import renderer, {act, type ReactTestRenderer} from 'react-test-renderer'; + +import { + createUploadClient, + type RequestRow, + type RequestsFilter, + type UploadClient, +} from 'react-native-background-upload'; +import {createFakeNative, type FakeNative} from '../../../src/testing'; + +// The package root builds a default client over the TurboModule at load. +jest.mock('../../../src/NativeRNFileUploader', () => ({ + __esModule: true, + default: jest + .requireActual( + '../../../src/testing', + ) + .createFakeNative(), +})); + +const row = (overrides: Partial = {}): RequestRow => ({ + id: 'a', + key: 'k', + vars: null, + state: 'queued', + bytesSent: 0, + totalBytes: 100, + attempts: 0, + updatedAt: 1, + ...overrides, +}); + +let native: FakeNative; +let client: UploadClient; + +beforeEach(() => { + native = createFakeNative(); + client = createUploadClient({native}); + client.configure({}); +}); + +type Seen = {renders: number; results: Array}; + +const Rows = ({ + seen, + filter, + progress = false, +}: { + seen: Seen; + filter?: RequestsFilter; + progress?: boolean; +}) => { + const rows = progress + ? client.useRequestsWithProgress(filter) + : client.useRequests(filter); + seen.renders += 1; + seen.results.push(rows); + return null; +}; + +const newSeen = (): Seen => ({renders: 0, results: []}); +const last = (seen: Seen) => seen.results[seen.results.length - 1]!; + +const mount = (element: React.ReactElement): ReactTestRenderer => { + let tree!: ReactTestRenderer; + act(() => { + tree = renderer.create(element); + }); + return tree; +}; + +it('reads native at the first render and at the first subscribe, not per hook', () => { + native.seedRows([row()]); + const getRequests = jest.spyOn(native, 'getRequests'); + const one = newSeen(); + const two = newSeen(); + mount( + <> + + + , + ); + // One read for the render, one check when the first hook subscribes. + expect(getRequests).toHaveBeenCalledTimes(2); + expect(last(one)).toEqual([row()]); + expect(last(two)).toEqual([row()]); + // The check found nothing new, so no hook rendered again. + expect(one.renders).toBe(1); +}); + +it('works under StrictMode, which subscribes twice', () => { + native.seedRows([row()]); + const seen = newSeen(); + const tree = mount( + + + , + ); + expect(native.listenerCount('state')).toBe(1); + expect(native.listenerCount('progress')).toBe(1); + act(() => native.emit.progress({id: 'a', bytesSent: 7, totalBytes: 100})); + expect(last(seen)[0]!.bytesSent).toBe(7); + act(() => native.emit.state(row({state: 'running', updatedAt: 2}))); + expect(last(seen)[0]!.state).toBe('running'); + act(() => tree.unmount()); + expect(native.listenerCount('state')).toBe(0); + expect(native.listenerCount('progress')).toBe(0); +}); + +it('follows a filter whose value changes', () => { + native.seedRows([row({id: 'a'}), row({id: 'b', key: 'other'})]); + const seen = newSeen(); + const tree = mount(); + expect(last(seen).map(r => r.id)).toEqual(['a']); + act(() => tree.update()); + expect(last(seen).map(r => r.id)).toEqual(['b']); + act(() => tree.update()); + expect(last(seen).map(r => r.id)).toEqual(['a']); + act(() => native.emit.state(row({id: 'b', key: 'other', state: 'running'}))); + expect(last(seen).map(r => r.id)).toEqual(['a']); +}); + +it('re-renders on a state event, and only the progress hook on progress', () => { + native.seedRows([row()]); + const plain = newSeen(); + const withProgress = newSeen(); + mount( + <> + + + , + ); + const plainRenders = plain.renders; + const progressRenders = withProgress.renders; + + act(() => native.emit.progress({id: 'a', bytesSent: 40, totalBytes: 100})); + expect(plain.renders).toBe(plainRenders); + expect(withProgress.renders).toBe(progressRenders + 1); + expect(last(withProgress)[0]!.bytesSent).toBe(40); + expect(last(plain)[0]!.bytesSent).toBe(0); + + act(() => native.emit.state(row({state: 'running', updatedAt: 2}))); + expect(plain.renders).toBe(plainRenders + 1); + expect(last(plain)[0]!.state).toBe('running'); + expect(last(withProgress)[0]!.state).toBe('running'); +}); + +it('returns the same filtered array across renders while its rows hold', () => { + native.seedRows([row({id: 'a'}), row({id: 'b'})]); + const seen = newSeen(); + // A new filter object each render, as an inline literal is. + const tree = mount(); + act(() => tree.update()); + expect(seen.renders).toBe(2); + expect(seen.results[1]).toBe(seen.results[0]); + + // A change to another row does not re-render this one. + act(() => native.emit.state(row({id: 'b', state: 'running'}))); + expect(seen.renders).toBe(2); + + act(() => native.emit.state(row({id: 'a', state: 'running'}))); + expect(seen.renders).toBe(3); + expect(last(seen)).toEqual([row({id: 'a', state: 'running'})]); +}); + +it('drops a completed row once its outcome is acknowledged', async () => { + const ping = client.define({ + key: 'ping', + request: () => ({url: 'https://api.test/ping'}), + }); + const seen = newSeen(); + mount(); + let id = ''; + await act(async () => { + ({id} = await ping.mutate()); + }); + expect(last(seen)).toEqual([expect.objectContaining({id, state: 'queued'})]); + await act(async () => { + await native.settle(id, {kind: 'completed'}); + }); + expect(last(seen)).toEqual([]); +}); + +it('drops a settled row when cancel() resolves, and hides cancelled rows', async () => { + native.seedRows([ + row({id: 'e', state: 'error'}), + row({id: 'c', state: 'cancelled'}), + ]); + const visible = newSeen(); + const all = newSeen(); + mount( + <> + + + , + ); + expect(last(visible).map(r => r.id)).toEqual(['e']); + expect(last(all).map(r => r.id)).toEqual(['e', 'c']); + await act(async () => { + await client.cancel('e'); + }); + expect(last(visible)).toEqual([]); + expect(last(all).map(r => r.id)).toEqual(['c']); +}); + +it('re-checks native when a hook mounts again', () => { + native.seedRows([row({id: 'a'})]); + const first = newSeen(); + const tree = mount(); + act(() => tree.unmount()); + // A change with no event while nothing is mounted. + native.seedRows([row({id: 'b'})]); + const second = newSeen(); + mount(); + expect(last(second).map(r => r.id)).toEqual(['a', 'b']); +}); + +it('types the vars of a definition', () => { + type Vars = {captureId: string}; + const upload = client.define({ + key: 'capture', + request: (_vars: Vars) => ({url: 'https://api.test/c'}), + }); + native.seedRows([ + row({id: 'x', key: 'capture', vars: {captureId: 'c1'}}), + row({id: 'y', key: 'other'}), + ]); + const ids: string[][] = []; + const Captures = () => { + const rows = upload.useRequests(); + ids.push(rows.map(r => r.vars.captureId)); + return null; + }; + mount(); + expect(ids[ids.length - 1]).toEqual(['c1']); +}); diff --git a/example/RNBGUExample/jest.config.js b/example/RNBGUExample/jest.config.js index 8eb675e9..3be97f22 100644 --- a/example/RNBGUExample/jest.config.js +++ b/example/RNBGUExample/jest.config.js @@ -1,3 +1,9 @@ module.exports = { preset: 'react-native', + // The library source in ../../src imports react. Without this it resolves + // the repo root's copy, and the test renderer's copy here is another React, + // so hooks fail. + moduleNameMapper: { + '^react$': '/node_modules/react', + }, }; diff --git a/src/__tests__/client.test.ts b/src/__tests__/client.test.ts index 6db7909c..2fcaf410 100644 --- a/src/__tests__/client.test.ts +++ b/src/__tests__/client.test.ts @@ -107,17 +107,21 @@ describe('client shape', () => { expect(Object.keys(client).sort()).toEqual( [ 'addListener', + 'addRequestsListener', 'android', 'cancel', 'chunkPlan', 'configure', 'define', + 'getCachedRequests', 'getRequests', 'pause', 'resume', 'setWifiOnly', 'takeDiagnostics', 'updateHeaders', + 'useRequests', + 'useRequestsWithProgress', ].sort(), ); expect(Object.keys(client.android)).toEqual(['addNotificationListener']); diff --git a/src/__tests__/delivery.test.ts b/src/__tests__/delivery.test.ts index ef82730f..4995c12c 100644 --- a/src/__tests__/delivery.test.ts +++ b/src/__tests__/delivery.test.ts @@ -56,15 +56,17 @@ const setup = ( }; const stateEvents: StateEvent[] = []; const warn = jest.fn(); + const onAcked = jest.fn(); const delivery = createDelivery({ native, lookup: (key) => definitions[key], emitState: (e) => stateEvents.push(e), + onAcked, warn, handlerWarningMs, }); const emit = (event: SettledEvent) => handlers.forEach((h) => h(event)); - return { ...delivery, native, emit, stateEvents, warn, handlers }; + return { ...delivery, native, emit, stateEvents, warn, handlers, onAcked }; }; describe('replay', () => { @@ -802,3 +804,93 @@ describe('slow handler warning', () => { expect(warn).not.toHaveBeenCalled(); }); }); + +describe('onAcked', () => { + it('reports each outcome after its ack resolves', async () => { + const ackGate = deferred(); + const { start, emit, native, onAcked } = setup({ + k: { key: 'k', request: jest.fn() }, + }); + native.ackEvents.mockReturnValueOnce(ackGate.promise); + start(); + await flush(); + emit(completed()); + await flush(); + expect(native.ackEvents).toHaveBeenCalledWith(['e1']); + expect(onAcked).not.toHaveBeenCalled(); + ackGate.resolve(); + await flush(); + expect(onAcked).toHaveBeenCalledTimes(1); + expect(onAcked).toHaveBeenCalledWith( + expect.objectContaining({ eventId: 'e1', id: 'u1', kind: 'completed' }), + ); + }); + + it('reports a cancelled outcome, which has no handler', async () => { + const { start, emit, onAcked } = setup({}); + start(); + await flush(); + emit({ + ...completed(), + kind: 'cancelled', + cancelReason: 'user', + state: 'cancelled', + } as SettledEvent); + await flush(); + expect(onAcked).toHaveBeenCalledWith( + expect.objectContaining({ id: 'u1', kind: 'cancelled' }), + ); + }); + + it('does not report an outcome whose ack failed', async () => { + const { start, emit, native, onAcked } = setup({ + k: { key: 'k', request: jest.fn() }, + }); + native.ackEvents.mockRejectedValueOnce(new Error('E_STORAGE')); + start(); + await flush(); + emit(completed()); + await flush(); + expect(onAcked).not.toHaveBeenCalled(); + }); + + it('does not report an outcome that was not acked', async () => { + const { start, emit, onAcked } = setup({ + k: { + key: 'k', + request: jest.fn(), + onSuccess: async () => { + throw new Error('no'); + }, + }, + }); + start(); + await flush(); + emit(completed()); + emit(completed({ eventId: 'e2', key: 'unknown' })); + await flush(); + expect(onAcked).not.toHaveBeenCalled(); + }); + + it('warns when onAcked throws, and keeps delivering', async () => { + const onSuccess = jest.fn(); + const { start, emit, native, onAcked, warn } = setup({ + k: { key: 'k', request: jest.fn(), onSuccess }, + }); + onAcked.mockImplementationOnce(() => { + throw new Error('boom'); + }); + start(); + await flush(); + emit(completed()); + emit(completed({ eventId: 'e2' })); + await flush(); + expect(native.ackEvents).toHaveBeenCalledTimes(2); + expect(onSuccess).toHaveBeenCalledTimes(2); + expect(onAcked).toHaveBeenCalledTimes(2); + expect(warn).toHaveBeenCalledWith( + expect.stringMatching(/after the ack of e1/), + expect.any(Error), + ); + }); +}); diff --git a/src/__tests__/requests.test.ts b/src/__tests__/requests.test.ts new file mode 100644 index 00000000..7391541d --- /dev/null +++ b/src/__tests__/requests.test.ts @@ -0,0 +1,1005 @@ +// The row cache behind useRequests(), tested through the client with the +// fake native. React is not rendered here (the root has no renderer); the +// example app's Jest renders the hooks with react-test-renderer. +jest.mock('react-native', () => { + const { createFakeNative: create } = jest.requireActual('../testing'); + const fake = create(); + return { + TurboModuleRegistry: { getEnforcing: () => fake, get: () => fake }, + }; +}); + +import { createUploadClient } from '../index'; +import { + createRequestCache, + createSelector, + jsonEqual, + MAX_FORGOTTEN, + rowEqual, + sharedSelector, +} from '../requests'; +import { createFakeNative, type FakeNative } from '../testing'; +import type { RequestRow } from '../types'; + +const row = (overrides: Partial = {}): RequestRow => ({ + id: 'a', + key: 'k', + vars: { n: 1 }, + state: 'queued', + bytesSent: 0, + totalBytes: 100, + attempts: 0, + updatedAt: 1, + ...overrides, +}); + +const flush = async (): Promise => { + for (let i = 0; i < 3; i++) { + await new Promise((resolve) => setImmediate(resolve)); + } +}; + +const setup = (native: FakeNative = createFakeNative()) => { + const warnings: string[] = []; + const client = createUploadClient({ native }); + const ping = client.define({ + key: 'ping', + request: (_vars: { n: number }) => ({ url: 'https://api.test/ping' }), + }); + client.configure({ onWarning: (message) => warnings.push(message) }); + const getRequests = jest.spyOn(native, 'getRequests'); + return { native, client, ping, warnings, getRequests }; +}; + +/** Subscribes the way a mounted hook does, and counts the notifications. */ +const listen = (client: ReturnType['client']) => { + const calls: Array = []; + const subscription = client.addRequestsListener((rows) => calls.push(rows)); + return { calls, subscription }; +}; + +describe('first subscribe', () => { + it('reads native once, however many listeners and reads follow', () => { + const { native, client, getRequests } = setup(); + native.seedRows([row({ id: 'a' }), row({ id: 'b' })]); + listen(client); + listen(client); + client.getCachedRequests(); + client.getCachedRequests({ key: 'k' }); + expect(getRequests).toHaveBeenCalledTimes(1); + expect(client.getCachedRequests().map((r) => r.id)).toEqual(['a', 'b']); + }); + + it('checks native at subscribe, so a row emitted after a render read is there', () => { + const { native, client } = setup(); + native.seedRows([row({ id: 'a' })]); + // A render reads before its effect subscribes. + expect(client.getCachedRequests().map((r) => r.id)).toEqual(['a']); + // Native queues a row and emits it while nothing listens: the event is lost. + native.seedRows([row({ id: 'b' })]); + native.emit.state(row({ id: 'b' })); + listen(client); + expect(client.getCachedRequests().map((r) => r.id)).toEqual(['a', 'b']); + }); +}); + +describe('state events', () => { + it('patch one row and keep the identity of the others', async () => { + const { native, client } = setup(); + native.seedRows([row({ id: 'a' }), row({ id: 'b' })]); + const { calls } = listen(client); + const before = client.getCachedRequests(); + native.emit.state(row({ id: 'a', state: 'running', updatedAt: 2 })); + const after = client.getCachedRequests(); + expect(after).not.toBe(before); + expect(after[0]).toEqual(row({ id: 'a', state: 'running', updatedAt: 2 })); + expect(after[1]).toBe(before[1]); + expect(calls).toHaveLength(1); + }); + + it('add a row that mutate() created', async () => { + const { client, ping } = setup(); + listen(client); + const { id } = await ping.mutate({ n: 1 }); + expect(client.getCachedRequests()).toEqual([ + expect.objectContaining({ id, key: 'ping', state: 'queued' }), + ]); + }); + + it('do not notify for a row that did not change', () => { + const { native, client } = setup(); + native.seedRows([row()]); + const { calls } = listen(client); + const before = client.getCachedRequests(); + native.emit.state(row()); + expect(client.getCachedRequests()).toBe(before); + expect(calls).toHaveLength(0); + }); + + it('take an unhandled-key row from native, not from the outcome', () => { + const { native, client } = setup(); + // A same-id mutate() reopened the entry after the outcome was journaled. + native.seedRows([row({ state: 'queued', updatedAt: 5 })]); + listen(client); + native.emit.state({ + ...row({ state: 'completed', updatedAt: 3 }), + reason: 'unhandled-key', + }); + expect(client.getCachedRequests()).toEqual([ + row({ state: 'queued', updatedAt: 5 }), + ]); + }); + + it('show the native row of an unhandled-key outcome, without the reason', () => { + const { native, client } = setup(); + native.seedRows([row({ state: 'running' })]); + listen(client); + native.seedRows([row({ state: 'completed', updatedAt: 2 })]); + native.emit.state({ + ...row({ state: 'completed', updatedAt: 2 }), + reason: 'unhandled-key', + }); + expect(client.getCachedRequests()).toEqual([ + row({ state: 'completed', updatedAt: 2 }), + ]); + }); +}); + +describe('progress', () => { + // The progress view is only reachable through the hook, so these tests + // build the cache over a client's events directly. + const progressSetup = () => { + const native = createFakeNative(); + const client = createUploadClient({ native }); + native.seedRows([row({ id: 'a' }), row({ id: 'b' })]); + const cache = createRequestCache({ + readNative: () => native.getRequests() as RequestRow[], + addListener: client.addListener, + warn: jest.fn(), + }); + return { native, cache }; + }; + + it('notifies only the progress view, and overlays the bytes', () => { + const { native, cache } = progressSetup(); + const rowCalls = jest.fn(); + const progressCalls = jest.fn(); + cache.subscribe(rowCalls); + cache.subscribeWithProgress(progressCalls); + const rows = cache.rows(); + const before = cache.rowsWithProgress(); + native.emit.progress({ id: 'a', bytesSent: 40, totalBytes: 100 }); + const after = cache.rowsWithProgress(); + expect(progressCalls).toHaveBeenCalledTimes(1); + expect(rowCalls).not.toHaveBeenCalled(); + expect(cache.rows()).toBe(rows); + expect(after).not.toBe(before); + expect(after[0]).toEqual(row({ id: 'a', bytesSent: 40 })); + expect(after[1]).toBe(before[1]); + // Reading again without a new event returns the same array. + expect(cache.rowsWithProgress()).toBe(after); + }); + + it('notifies both views on a state event', () => { + const { native, cache } = progressSetup(); + const rowCalls = jest.fn(); + const progressCalls = jest.fn(); + cache.subscribe(rowCalls); + cache.subscribeWithProgress(progressCalls); + native.emit.state(row({ id: 'a', state: 'running', updatedAt: 2 })); + expect(rowCalls).toHaveBeenCalledTimes(1); + expect(progressCalls).toHaveBeenCalledTimes(1); + }); + + it('gives way to the bytes of the next state event', () => { + const { native, cache } = progressSetup(); + cache.subscribeWithProgress(() => {}); + native.emit.progress({ id: 'a', bytesSent: 40, totalBytes: 100 }); + native.emit.state(row({ state: 'running', bytesSent: 10, updatedAt: 2 })); + expect(cache.rowsWithProgress()[0]!.bytesSent).toBe(10); + }); + + it('ignores progress for a row it does not hold', () => { + const { native, cache } = progressSetup(); + const progressCalls = jest.fn(); + cache.subscribeWithProgress(progressCalls); + native.emit.progress({ id: 'zzz', bytesSent: 1, totalBytes: 2 }); + expect(progressCalls).not.toHaveBeenCalled(); + }); +}); + +describe('rows native forgets with no event', () => { + it('drops a completed row when its ack resolves', async () => { + const { native, client, ping } = setup(); + const { calls } = listen(client); + const { id } = await ping.mutate({ n: 1 }); + // settle() resolves after the client handled the ack, so the cache is + // already up to date here. + await native.settle(id, { kind: 'completed' }); + expect(native.getRequests()).toEqual([]); + expect(client.getCachedRequests({ includeCancelled: true })).toEqual([]); + expect(calls[calls.length - 1]).toEqual([]); + }); + + it('keeps a completed row whose ack failed', async () => { + const native = createFakeNative({ ackTimeoutMs: 50 }); + const { client, ping, warnings } = setup(native); + listen(client); + const { id } = await ping.mutate({ n: 1 }); + native.failNext('ackEvents', 'E_STORAGE'); + await expect(native.settle(id, { kind: 'completed' })).rejects.toThrow( + /not acknowledged/, + ); + expect(client.getCachedRequests()).toEqual([ + expect.objectContaining({ id, state: 'completed' }), + ]); + expect(warnings).toEqual([expect.stringMatching(/ackEvents failed/)]); + }); + + it('keeps a completed row whose key has no definition', async () => { + const native = createFakeNative({ ackTimeoutMs: 50 }); + const { client } = setup(native); + native.seedRows([row({ id: 'x', key: 'gone', state: 'running' })]); + listen(client); + await expect(native.settle('x', { kind: 'completed' })).rejects.toThrow( + /not acknowledged/, + ); + expect(client.getCachedRequests()).toEqual([ + expect.objectContaining({ id: 'x', state: 'completed' }), + ]); + }); + + it('drops a cancelled live row when its ack resolves', async () => { + const { native, client, ping } = setup(); + listen(client); + const { id } = await ping.mutate({ n: 1 }); + // The fake, as Android, emits the cancelled state before cancel() + // resolves. cancel() itself does not drop the row. + await client.cancel(id); + expect(client.getCachedRequests({ includeCancelled: true })).toEqual([ + expect.objectContaining({ id, state: 'cancelled' }), + ]); + await flush(); + expect(native.ackedEventIds).toHaveLength(1); + expect(native.getRequests()).toEqual([]); + expect(client.getCachedRequests({ includeCancelled: true })).toEqual([]); + }); + + it('keeps a cancelled live row whose ack failed, as native does', async () => { + const { native, client, ping, warnings } = setup(); + listen(client); + const { id } = await ping.mutate({ n: 1 }); + native.failNext('ackEvents', 'E_STORAGE'); + await client.cancel(id); + await flush(); + expect(native.getRequests()).toEqual([ + expect.objectContaining({ id, state: 'cancelled' }), + ]); + expect(client.getCachedRequests({ includeCancelled: true })).toEqual([ + expect.objectContaining({ id, state: 'cancelled' }), + ]); + expect(warnings).toEqual([expect.stringMatching(/ackEvents failed/)]); + }); + + it('drops a row whose outcome was acked before its last state event', async () => { + // Android's order: the outcome, then the last state event. Here the + // client gets the outcome and acks it before that state event arrives. + const fake = createFakeNative(); + let release: (() => void) | undefined; + const native: FakeNative = { + ...fake, + onState: (handler) => + fake.onState((row) => { + if ((row as RequestRow).state === 'completed') { + release = () => void handler(row); + return; + } + return handler(row); + }), + }; + const { client, ping } = setup(native); + const { calls } = listen(client); + const { id } = await ping.mutate({ n: 1 }); + fake.emit.state(row({ id, key: 'ping', state: 'running', updatedAt: 2 })); + await fake.settle(id, { kind: 'completed' }); + expect(fake.getRequests()).toEqual([]); + expect(client.getCachedRequests()).toEqual([]); + const seen = calls.length; + release?.(); + expect(client.getCachedRequests()).toEqual([]); + expect(calls).toHaveLength(seen); + }); + + it('drops a settled row when cancel() resolves', async () => { + const { native, client } = setup(); + native.seedRows([row({ id: 'e', state: 'error' })]); + const { calls } = listen(client); + const states = jest.fn(); + client.addListener('state', states); + await client.cancel('e'); + expect(states).not.toHaveBeenCalled(); + expect(client.getCachedRequests()).toEqual([]); + expect(calls).toEqual([[]]); + }); + + it('keeps the row when cancel() rejects', async () => { + const { native, client } = setup(); + native.seedRows([row({ id: 'e', state: 'error' })]); + listen(client); + native.failNext('cancel', 'E_STORAGE'); + await expect(client.cancel('e')).rejects.toThrow('E_STORAGE'); + expect(client.getCachedRequests()).toHaveLength(1); + }); + + it('keeps an error row after its ack, until cancel()', async () => { + const { native, client, ping } = setup(); + listen(client); + const { id } = await ping.mutate({ n: 1 }); + await native.settle(id, { kind: 'error', error: { errorKind: 'http' } }); + expect(client.getCachedRequests()).toEqual([ + expect.objectContaining({ id, state: 'error' }), + ]); + await client.cancel(id); + expect(client.getCachedRequests()).toEqual([]); + }); + + it('replaces an error row on a same-id mutate()', async () => { + const { native, client, ping } = setup(); + listen(client); + const { id } = await ping.mutate({ n: 1 }); + await native.settle(id, { kind: 'error', error: { errorKind: 'http' } }); + await ping.mutate({ n: 2 }, { id }); + expect(client.getCachedRequests()).toEqual([ + expect.objectContaining({ id, state: 'queued', vars: { n: 2 } }), + ]); + }); +}); + +describe('the ack of an outcome whose row moved on', () => { + const stub = (nativeRows: RequestRow[]) => { + const listeners: Record void> = {}; + const readNative = jest.fn(() => nativeRows.map((r) => ({ ...r }))); + const cache = createRequestCache({ + readNative, + addListener: ((event: string, listener: (e: never) => void) => { + listeners[event] = listener; + return { remove: () => {} }; + }) as never, + warn: jest.fn(), + }); + return { cache, readNative, listeners }; + }; + + it('asks native when the row is not in the outcome state, and keeps it', async () => { + // A cancelled entry reopened by a same-id mutate(): the old outcome's + // ack does not forget it. + const reopened = row({ id: 'a', state: 'queued', updatedAt: 5 }); + const { cache, readNative } = stub([reopened]); + cache.subscribe(() => {}); + await Promise.resolve(); + readNative.mockClear(); + cache.acked({ id: 'a', kind: 'cancelled' }); + expect(readNative).toHaveBeenCalledTimes(1); + expect(cache.rows()).toEqual([reopened]); + }); + + it('drops the row when native no longer has it', async () => { + const nativeRows = [row({ id: 'a', state: 'running' })]; + const { cache } = stub(nativeRows); + cache.subscribe(() => {}); + nativeRows.length = 0; + cache.acked({ id: 'a', kind: 'completed' }); + expect(cache.rows()).toEqual([]); + }); + + it('does nothing for an error outcome on an error row', () => { + const { cache, readNative } = stub([row({ id: 'a', state: 'error' })]); + cache.subscribe(() => {}); + readNative.mockClear(); + cache.acked({ id: 'a', kind: 'error' }); + expect(readNative).not.toHaveBeenCalled(); + expect(cache.rows()).toHaveLength(1); + }); + + it('asks native for an error outcome on a row still shown live', () => { + // iOS can settle a live row during the ack with no event. + const nativeRows = [row({ id: 'a', state: 'running' })]; + const { cache, readNative } = stub(nativeRows); + cache.subscribe(() => {}); + nativeRows[0] = row({ id: 'a', state: 'error', updatedAt: 2 }); + readNative.mockClear(); + cache.acked({ id: 'a', kind: 'error' }); + expect(readNative).toHaveBeenCalledTimes(1); + expect(cache.rows()).toEqual([row({ id: 'a', state: 'error', updatedAt: 2 })]); + }); +}); + +describe('a state event that arrives after the cache removed its row', () => { + // Android emits a run's outcome, then its last state event. The ack of the + // outcome can forget the row and resolve before that state event reaches + // JS. This stub drives the events by hand to get that order. + const stub = (nativeRows: RequestRow[]) => { + let onState: ((e: unknown) => void) | undefined; + let failReads = false; + const readNative = jest.fn(() => { + if (failReads) { + throw new Error('E_BOOM'); + } + return nativeRows.map((r) => ({ ...r })); + }); + const cache = createRequestCache({ + readNative, + addListener: ((event: string, listener: (e: never) => void) => { + if (event === 'state') { + onState = listener as (e: unknown) => void; + } + return { + remove: () => { + if (event === 'state') { + onState = undefined; + } + }, + }; + }) as never, + warn: jest.fn(), + }); + const emitState = (r: RequestRow) => onState?.(r); + const setFailReads = (fail: boolean) => { + failReads = fail; + }; + return { cache, readNative, emitState, setFailReads }; + }; + + /** Ack while the row is running and native has forgotten it. */ + const ackedWhileRunning = () => { + const nativeRows = [row({ id: 'a', state: 'running', updatedAt: 2 })]; + const s = stub(nativeRows); + const listener = jest.fn(); + const subscription = s.cache.subscribe(listener); + nativeRows.length = 0; + s.cache.acked({ id: 'a', kind: 'completed' }); + expect(s.cache.rows()).toEqual([]); + expect(listener).toHaveBeenCalledTimes(1); + listener.mockClear(); + s.readNative.mockClear(); + return { ...s, nativeRows, listener, subscription }; + }; + + it('does not bring the row back on a late completed event', () => { + const { cache, emitState, listener, readNative } = ackedWhileRunning(); + emitState(row({ id: 'a', state: 'completed', updatedAt: 3 })); + expect(cache.rows()).toEqual([]); + expect(listener).not.toHaveBeenCalled(); + // It asked native once instead of trusting the event. + expect(readNative).toHaveBeenCalledTimes(1); + }); + + it('does not bring the row back on a late running event, then completed', () => { + const { cache, emitState, listener } = ackedWhileRunning(); + emitState(row({ id: 'a', state: 'running', updatedAt: 2, bytesSent: 50 })); + emitState(row({ id: 'a', state: 'completed', updatedAt: 3 })); + expect(cache.rows()).toEqual([]); + expect(listener).not.toHaveBeenCalled(); + }); + + it('shows the row of a same-id mutate() after the ack', () => { + const { cache, emitState, listener, nativeRows, readNative } = + ackedWhileRunning(); + const reopened = row({ + id: 'a', + state: 'queued', + vars: { n: 2 }, + updatedAt: 9, + }); + nativeRows.push(reopened); + emitState(reopened); + expect(cache.rows()).toEqual([reopened]); + expect(listener).toHaveBeenCalledTimes(1); + // Native showed the row again, so its next events apply without a read. + readNative.mockClear(); + emitState( + row({ id: 'a', state: 'running', vars: { n: 2 }, updatedAt: 10 }), + ); + expect(readNative).not.toHaveBeenCalled(); + expect(cache.rows()).toEqual([ + row({ id: 'a', state: 'running', vars: { n: 2 }, updatedAt: 10 }), + ]); + }); + + it('does not ask native for the first event of a new id', () => { + const { cache, emitState, readNative } = ackedWhileRunning(); + emitState(row({ id: 'b' })); + expect(readNative).not.toHaveBeenCalled(); + expect(cache.rows()).toEqual([row({ id: 'b' })]); + }); + + it('keeps a bounded number of forgotten ids, oldest out first', () => { + const { cache, emitState, readNative } = stub([]); + cache.subscribe(() => {}); + for (let i = 0; i <= MAX_FORGOTTEN; i++) { + cache.acked({ id: `r${i}`, kind: 'completed' }); + } + readNative.mockClear(); + // r0 was pushed out by the last ack, so a live event applies as a new + // row with no read. + emitState(row({ id: 'r0', state: 'running' })); + expect(readNative).not.toHaveBeenCalled(); + // r1 is still kept, so its late event is checked and dropped. + emitState(row({ id: 'r1', state: 'running' })); + expect(readNative).toHaveBeenCalledTimes(1); + expect(cache.rows().map((r) => r.id)).toEqual(['r0']); + }); + + it('does not bring back an id pushed out of the set on a late settled event', () => { + const { cache, emitState, listener, readNative } = ackedWhileRunning(); + for (let i = 0; i < MAX_FORGOTTEN; i++) { + cache.acked({ id: `o${i}`, kind: 'completed' }); + } + listener.mockClear(); + readNative.mockClear(); + // 'a' is out of the set, but a settled state for an id the cache does + // not hold is checked too. + emitState(row({ id: 'a', state: 'completed', updatedAt: 3 })); + expect(readNative).toHaveBeenCalledTimes(1); + expect(cache.rows()).toEqual([]); + expect(listener).not.toHaveBeenCalled(); + }); + + it('shows the live events of a reused id when the native read fails', () => { + const { cache, emitState, nativeRows, setFailReads } = ackedWhileRunning(); + setFailReads(true); + nativeRows.push(row({ id: 'a', state: 'queued', updatedAt: 9 })); + emitState(row({ id: 'a', state: 'queued', updatedAt: 9 })); + expect(cache.rows()).toEqual([ + row({ id: 'a', state: 'queued', updatedAt: 9 }), + ]); + emitState(row({ id: 'a', state: 'running', updatedAt: 10 })); + // The row is shown now, so its settled state applies too. + emitState(row({ id: 'a', state: 'error', updatedAt: 11 })); + expect(cache.rows()).toEqual([ + row({ id: 'a', state: 'error', updatedAt: 11 }), + ]); + }); + + it('drops a late settled event of a forgotten id when the native read fails', () => { + const { cache, emitState, setFailReads } = ackedWhileRunning(); + setFailReads(true); + emitState(row({ id: 'a', state: 'completed', updatedAt: 3 })); + expect(cache.rows()).toEqual([]); + }); + + it('on a full read, stops checking ids native shows and keeps the others', () => { + const { cache, emitState, readNative, nativeRows, subscription } = + ackedWhileRunning(); + cache.acked({ id: 'b', kind: 'completed' }); + // A component replaces the last listener in one commit: a full read. + subscription(); + const reopened = row({ id: 'b', state: 'queued', updatedAt: 9 }); + nativeRows.push(reopened); + cache.subscribe(() => {}); + expect(cache.rows()).toEqual([reopened]); + readNative.mockClear(); + // 'b' is shown again, so its event applies without a read. + emitState(row({ id: 'b', state: 'running', updatedAt: 10 })); + expect(readNative).not.toHaveBeenCalled(); + // The late event of 'a', queued before the new subscribe, is checked. + emitState(row({ id: 'a', state: 'completed', updatedAt: 3 })); + expect(readNative).toHaveBeenCalledTimes(1); + expect(cache.rows().map((r) => r.id)).toEqual(['b']); + }); + + it('does not bring the row back after an ack while nothing listened', () => { + const nativeRows = [row({ id: 'a', state: 'running' })]; + const { cache, emitState } = stub(nativeRows); + const first = cache.subscribe(() => {}); + first(); + nativeRows.length = 0; + cache.acked({ id: 'a', kind: 'completed' }); + cache.subscribe(() => {}); + emitState(row({ id: 'a', state: 'completed', updatedAt: 3 })); + expect(cache.rows()).toEqual([]); + }); + + it('does not bring back a row that cancel() forgot while it was shown live', () => { + // Native settled the row as an error, then cancel() forgot it before + // the error state event reached JS. + const nativeRows = [row({ id: 'a', state: 'running' })]; + const { cache, emitState, readNative } = stub(nativeRows); + cache.subscribe(() => {}); + nativeRows.length = 0; + readNative.mockClear(); + cache.cancelled('a'); + expect(readNative).toHaveBeenCalledTimes(1); + expect(cache.rows()).toEqual([]); + emitState(row({ id: 'a', state: 'error', updatedAt: 3 })); + expect(cache.rows()).toEqual([]); + }); + + it('keeps a live row that cancel() moved to cancelled until its ack', () => { + // Android's order: the cancelled state event arrives before cancel() + // resolves, and native keeps the row until the ack. + const cancelledRow = row({ id: 'a', state: 'cancelled', updatedAt: 3 }); + const nativeRows = [row({ id: 'a', state: 'running' })]; + const { cache, emitState } = stub(nativeRows); + cache.subscribe(() => {}); + nativeRows[0] = cancelledRow; + emitState(cancelledRow); + cache.cancelled('a'); + expect(cache.rows()).toEqual([cancelledRow]); + nativeRows.length = 0; + cache.acked({ id: 'a', kind: 'cancelled' }); + expect(cache.rows()).toEqual([]); + }); + + it('drops a cancelled row that a second cancel() forgot', () => { + const nativeRows = [row({ id: 'a', state: 'cancelled', updatedAt: 3 })]; + const { cache } = stub(nativeRows); + cache.subscribe(() => {}); + nativeRows.length = 0; + cache.cancelled('a'); + expect(cache.rows()).toEqual([]); + }); + + it('shows a same-id mutate() after cancel() of an id it did not hold', () => { + const nativeRows: RequestRow[] = []; + const { cache, emitState, readNative } = stub(nativeRows); + cache.subscribe(() => {}); + cache.cancelled('a'); + const reopened = row({ id: 'a', state: 'queued', updatedAt: 9 }); + nativeRows.push(reopened); + readNative.mockClear(); + emitState(reopened); + expect(readNative).toHaveBeenCalledTimes(1); + expect(cache.rows()).toEqual([reopened]); + }); +}); + +describe('unsubscribe and subscribe again', () => { + it('keeps the rows, then checks them against native', async () => { + const { native, client, getRequests } = setup(); + native.seedRows([row({ id: 'a' })]); + const first = listen(client); + const kept = client.getCachedRequests(); + first.subscription.remove(); + await flush(); + // A change with no event while nobody listens. + native.seedRows([row({ id: 'b' })]); + getRequests.mockClear(); + const second = listen(client); + expect(getRequests).toHaveBeenCalledTimes(1); + expect(client.getCachedRequests().map((r) => r.id)).toEqual(['a', 'b']); + // Row 'a' did not change, so it keeps its identity. + expect(client.getCachedRequests()[0]).toBe(kept[0]); + // No call while the listener registers; it reads the rows itself. + expect(second.calls).toHaveLength(0); + }); + + it('keeps the same array when native has not changed', async () => { + const { native, client } = setup(); + native.seedRows([row()]); + const first = listen(client); + const kept = client.getCachedRequests(); + first.subscription.remove(); + await flush(); + const second = listen(client); + expect(client.getCachedRequests()).toBe(kept); + expect(second.calls).toHaveLength(0); + }); + + it('stops listening to native when the last listener leaves', () => { + const { native, client } = setup(); + const before = native.listenerCount('state'); + const one = listen(client); + const two = listen(client); + expect(native.listenerCount('state')).toBe(before + 1); + one.subscription.remove(); + expect(native.listenerCount('state')).toBe(before + 1); + two.subscription.remove(); + expect(native.listenerCount('state')).toBe(before); + }); + + it('reads native on every getCachedRequests() while nobody listens', () => { + const { native, client, getRequests } = setup(); + native.seedRows([row({ id: 'a' })]); + expect(client.getCachedRequests().map((r) => r.id)).toEqual(['a']); + native.seedRows([row({ id: 'b' })]); + expect(client.getCachedRequests().map((r) => r.id)).toEqual(['a', 'b']); + expect(getRequests).toHaveBeenCalledTimes(2); + }); + + it('warns and keeps the rows when the native read throws', async () => { + const { native, client, warnings } = setup(); + native.seedRows([row()]); + const first = listen(client); + first.subscription.remove(); + await flush(); + native.failNext('getRequests', 'E_BOOM'); + listen(client); + expect(client.getCachedRequests()).toHaveLength(1); + expect(warnings).toEqual([expect.stringMatching(/getRequests/)]); + }); +}); + +describe('two clients', () => { + it('keep separate caches', async () => { + const one = setup(); + const two = setup(); + one.native.seedRows([row({ id: 'a' })]); + two.native.seedRows([row({ id: 'b' })]); + listen(one.client); + listen(two.client); + one.native.emit.state(row({ id: 'c' })); + expect(one.client.getCachedRequests().map((r) => r.id)).toEqual([ + 'a', + 'c', + ]); + expect(two.client.getCachedRequests().map((r) => r.id)).toEqual(['b']); + }); +}); + +describe('filters', () => { + it('leave out cancelled rows unless includeCancelled is set', () => { + const { native, client } = setup(); + native.seedRows([row({ id: 'a' }), row({ id: 'c', state: 'cancelled' })]); + listen(client); + expect(client.getCachedRequests().map((r) => r.id)).toEqual(['a']); + expect( + client.getCachedRequests({ includeCancelled: true }).map((r) => r.id), + ).toEqual(['a', 'c']); + }); + + it('match key and id', () => { + const { native, client } = setup(); + native.seedRows([ + row({ id: 'a', key: 'k1' }), + row({ id: 'b', key: 'k2' }), + row({ id: 'c', key: 'k2' }), + ]); + listen(client); + expect(client.getCachedRequests({ key: 'k2' }).map((r) => r.id)).toEqual([ + 'b', + 'c', + ]); + expect( + client.getCachedRequests({ key: 'k2', id: 'c' }).map((r) => r.id), + ).toEqual(['c']); + }); + + it('return the same array while nothing changed', () => { + const { native, client } = setup(); + native.seedRows([row()]); + listen(client); + expect(client.getCachedRequests()).toBe(client.getCachedRequests()); + expect(client.getCachedRequests({ includeCancelled: true })).toBe( + client.getCachedRequests({ includeCancelled: true }), + ); + }); +}); + +describe('createSelector', () => { + it('returns the same array for the same rows', () => { + const rows = [row({ id: 'a' }), row({ id: 'b' })]; + const select = createSelector(() => rows, { id: 'a' }); + expect(select()).toBe(select()); + }); + + it('keeps its array when only rows it leaves out change', () => { + let rows = [row({ id: 'a' }), row({ id: 'b' })]; + const select = createSelector(() => rows, { id: 'a' }); + const first = select(); + rows = [rows[0]!, row({ id: 'b', state: 'running' })]; + expect(select()).toBe(first); + rows = [row({ id: 'a', state: 'running' }), rows[1]!]; + expect(select()).not.toBe(first); + expect(select()).toEqual([row({ id: 'a', state: 'running' })]); + }); + + it('keeps its array when a hidden cancelled row changes', () => { + let rows = [row({ id: 'a' }), row({ id: 'c', state: 'cancelled' })]; + const select = createSelector(() => rows, {}); + const first = select(); + expect(first.map((r) => r.id)).toEqual(['a']); + rows = [rows[0]!]; + expect(select()).toBe(first); + }); +}); + +describe('addRequestsListener', () => { + it('is not called while it registers, even when the check finds a change', async () => { + const { native, client } = setup(); + native.seedRows([row({ id: 'a' })]); + const first = listen(client); + first.subscription.remove(); + native.seedRows([row({ id: 'b' })]); + const listener = jest.fn(); + client.addRequestsListener(listener); + expect(listener).not.toHaveBeenCalled(); + expect(client.getCachedRequests().map((r) => r.id)).toEqual(['a', 'b']); + }); + + it('is not called when only a hidden cancelled row changes', () => { + const { native, client } = setup(); + native.seedRows([row({ id: 'a' }), row({ id: 'c', state: 'cancelled' })]); + const { calls } = listen(client); + native.emit.state(row({ id: 'c', state: 'cancelled', updatedAt: 9 })); + expect(calls).toHaveLength(0); + native.emit.state(row({ id: 'a', state: 'running' })); + expect(calls).toHaveLength(1); + }); + + it('may call the client while the cache drops a row after an ack', async () => { + const { native, client, ping, warnings } = setup(); + const { id: first } = await ping.mutate({ n: 1 }); + const { id: second } = await ping.mutate({ n: 2 }); + await native.settle(second, { kind: 'error', error: { errorKind: 'http' } }); + const seen: string[][] = []; + const cancels: Array> = []; + client.addRequestsListener((rows) => { + seen.push(rows.map((r) => r.id)); + // Reads native and starts a cancel from inside the notification. + client.getRequests(); + if (cancels.length === 0 && !rows.some((r) => r.id === first)) { + cancels.push(client.cancel(second)); + } + }); + await native.settle(first, { kind: 'completed' }); + await Promise.all(cancels); + expect(cancels).toHaveLength(1); + expect(client.getCachedRequests()).toEqual([]); + expect(native.getRequests()).toEqual([]); + // The completed state event, the drop after the ack, then the cancel. + expect(seen.slice(-2)).toEqual([[second], []]); + expect(warnings).toEqual([]); + }); +}); + +describe('selector work', () => { + /** An array that counts reads of its elements. */ + const counted = (rows: RequestRow[]) => { + const reads = { count: 0 }; + const proxy = new Proxy(rows, { + get: (target, prop, receiver) => { + if (typeof prop === 'string' && /^\d+$/.test(prop)) { + reads.count++; + } + return Reflect.get(target, prop, receiver); + }, + }); + return { proxy, reads }; + }; + + it('looks an id up in one index per rows array, without walking the rows each time', () => { + const many = Array.from({ length: 100 }, (_, i) => row({ id: `r${i}` })); + const { proxy, reads } = counted(many); + const selectors = Array.from({ length: 50 }, (_, i) => + createSelector(() => proxy, { id: `r${i}` }), + ); + const results = selectors.map((select) => select()); + // One index build reads each row once; 50 selectors share it. + expect(reads.count).toBe(100); + expect(results[7]).toEqual([many[7]]); + expect(selectors[7]!()).toBe(results[7]); + }); + + it('keeps an id result when a new array holds the same row object', () => { + const rows = [row({ id: 'a' }), row({ id: 'b' })]; + let current = rows; + const select = createSelector(() => current, { id: 'a' }); + const first = select(); + current = [rows[0]!, row({ id: 'b', state: 'running' })]; + expect(select()).toBe(first); + current = [row({ id: 'a', state: 'cancelled' }), current[1]!]; + expect(select()).toEqual([]); + }); + + it('shares one selector per source and filter without an id', () => { + const rows = [row({ id: 'a', key: 'k1' }), row({ id: 'b', key: 'k2' })]; + const source = () => rows; + expect(sharedSelector(source, {})).toBe(sharedSelector(source, {})); + expect(sharedSelector(source, { key: 'k1' })).toBe( + sharedSelector(source, { key: 'k1' }), + ); + expect(sharedSelector(source, { key: 'k1' })).not.toBe( + sharedSelector(source, {}), + ); + expect(sharedSelector(source, { includeCancelled: true })).not.toBe( + sharedSelector(source, {}), + ); + expect(sharedSelector(source, { key: 'k1' })().map((r) => r.id)).toEqual([ + 'a', + ]); + }); +}); + +describe('jsonEqual', () => { + it('compares JSON values deeply, with undefined fields as absent', () => { + expect(jsonEqual({ a: [1, { b: null }] }, { a: [1, { b: null }] })).toBe( + true, + ); + expect(jsonEqual({ a: 1, b: undefined }, { a: 1 })).toBe(true); + expect(jsonEqual({ a: 1 }, { a: 1, b: 2 })).toBe(false); + expect(jsonEqual({ a: 1, c: 2 }, { a: 1, b: 2 })).toBe(false); + expect(jsonEqual([1, 2], [1, 2, 3])).toBe(false); + expect(jsonEqual([], {})).toBe(false); + expect(jsonEqual(null, {})).toBe(false); + expect(jsonEqual('1', 1)).toBe(false); + }); + + it('finds a changed vars field when every scalar field is the same', () => { + expect( + rowEqual(row({ vars: { n: 1 } }), row({ vars: { n: 2 } })), + ).toBe(false); + expect(rowEqual(row({ vars: { n: 1 } }), row({ vars: { n: 1 } }))).toBe( + true, + ); + }); +}); + +describe('guards', () => { + it('warns instead of throwing out of a native state callback', () => { + const native = createFakeNative(); + const client = createUploadClient({ native }); + native.seedRows([row()]); + const warn = jest.fn(); + const cache = createRequestCache({ + readNative: () => native.getRequests() as RequestRow[], + addListener: client.addListener, + warn, + }); + // The cache's listener is registered first, so a throw in it would stop + // the app's listener below. + cache.subscribe(() => { + throw new Error('boom'); + }); + cache.subscribeWithProgress(() => { + throw new Error('boom'); + }); + const appListener = jest.fn(); + client.addListener('state', appListener); + client.addListener('progress', appListener); + expect(() => + native.emit.state(row({ state: 'running', updatedAt: 2 })), + ).not.toThrow(); + expect(() => + native.emit.progress({ id: 'a', bytesSent: 1, totalBytes: 100 }), + ).not.toThrow(); + expect(appListener).toHaveBeenCalledTimes(2); + expect(warn).toHaveBeenCalledWith( + expect.stringMatching(/from a state event threw/), + expect.any(Error), + ); + expect(warn).toHaveBeenCalledWith( + expect.stringMatching(/from a progress event threw/), + expect.any(Error), + ); + }); +}); + +describe('with nothing listening', () => { + it('drops the rows at an ack instead of reading native', async () => { + const { native, client, ping, getRequests } = setup(); + const { id } = await ping.mutate({ n: 1 }); + // A read with no listener fills the cache, then nobody listens. + client.getCachedRequests(); + native.seedRows([row({ id, key: 'ping', state: 'running' })]); + getRequests.mockClear(); + // The cached row is queued, the outcome completed: with a listener this + // would read native. + await native.settle(id, { kind: 'completed' }); + expect(getRequests).not.toHaveBeenCalled(); + // The next read takes native's rows. + expect(client.getCachedRequests()).toEqual([]); + }); + + it('drops the rows at cancel() too, and the next subscribe reads native', async () => { + const { native, client, getRequests } = setup(); + native.seedRows([row({ id: 'e', state: 'error' })]); + client.getCachedRequests(); + getRequests.mockClear(); + await client.cancel('e'); + expect(getRequests).not.toHaveBeenCalled(); + listen(client); + expect(getRequests).toHaveBeenCalledTimes(1); + expect(client.getCachedRequests()).toEqual([]); + }); +}); diff --git a/src/__typetests__/requests.ts b/src/__typetests__/requests.ts new file mode 100644 index 00000000..5f5b6658 --- /dev/null +++ b/src/__typetests__/requests.ts @@ -0,0 +1,82 @@ +// Type-level tests for the row hooks. tsc compiles this file with the library +// (yarn typecheck); nothing here runs. +import type { + DefinedRequestRow, + RequestRow, + RequestsFilter, + UploadClient, +} from '../types'; + +type Equal = (() => T extends A ? 1 : 2) extends () => T extends B + ? 1 + : 2 + ? true + : false; +const expectType = (_actual: Expected): void => {}; +const assertEqual = ( + _proof: Equal extends true ? true : never, +) => {}; + +declare const client: UploadClient; + +// The client hooks take the getRequests() filter plus includeCancelled. +const rows = client.useRequests(); +assertEqual(true); +client.useRequests({ key: 'k', id: 'i', includeCancelled: true }); +client.useRequestsWithProgress({ key: 'k' }); +// @ts-expect-error unknown filter field +client.useRequests({ keys: ['k'] }); +// @ts-expect-error includeCancelled is a boolean +client.useRequests({ includeCancelled: 'yes' }); + +// The array is shared with every other reader, so it is read-only. +// @ts-expect-error push does not exist on a readonly array +rows.push(rows[0]!); + +// Outside React. +expectType(client.getCachedRequests({ key: 'k' })); +const subscription = client.addRequestsListener((current) => { + expectType(current); +}); +subscription.remove(); +const filter: RequestsFilter = { includeCancelled: false }; +client.getCachedRequests(filter); + +// A definition's hooks type the vars, and take no key. +type CaptureVars = { captureId: string; index: number }; +const capture = client.define({ + key: 'capture.upload', + request: ({ captureId }: CaptureVars) => ({ + url: `https://api/captures/${captureId}`, + }), +}); +const captureRows = capture.useRequests(); +assertEqual[]>( + true, +); +const first = captureRows[0]; +if (first) { + assertEqual(true); + assertEqual(true); +} +capture.useRequests({ id: 'x', includeCancelled: true }); +capture.useRequestsWithProgress({ id: 'x' }); +// @ts-expect-error the key is the definition's +capture.useRequests({ key: 'other' }); + +// A definition with a response parser types its rows the same way. +const parsed = client.define({ + key: 'note.add', + request: (_vars: { noteId: string }) => ({ url: 'https://api/notes' }), + response: (raw) => raw as { id: string }, +}); +expectType[]>( + parsed.useRequestsWithProgress(), +); + +// A definition without vars has rows whose vars are null. +const ping = client.define({ + key: 'ping', + request: () => ({ url: 'https://api/ping' }), +}); +expectType[]>(ping.useRequests()); diff --git a/src/delivery.ts b/src/delivery.ts index 685145d1..eda1c0cc 100644 --- a/src/delivery.ts +++ b/src/delivery.ts @@ -49,6 +49,12 @@ type DeliveryDeps = { lookup: (key: string) => AnyDefinition | undefined; /** Feeds the client's `state` listeners. */ emitState: (event: StateEvent) => void; + /** + * Called after the ack of an outcome resolved, so the client's row cache + * can drop a row that native forgot with no event. Not called when the ack + * failed. A throw is a warning and changes nothing in delivery. + */ + onAcked?: (event: SettledEvent) => void; warn?: (message: string, detail?: unknown) => void; handlerWarningMs?: number; }; @@ -76,6 +82,7 @@ export const createDelivery = ({ native, lookup, emitState, + onAcked, warn = console.warn, handlerWarningMs = HANDLER_WARNING_MS, }: DeliveryDeps): Delivery => { @@ -105,11 +112,26 @@ export const createDelivery = ({ }); }; - const ack = async (eventId: string): Promise => { + const ack = async (event: SettledEvent): Promise => { + const { eventId } = event; try { await native.ackEvents([eventId]); } catch (e) { warn(`delivery: ackEvents failed for ${eventId}`, e); + return; + } + // Display only: the ack has already landed, and a throw here is caught. + // This runs one promise turn after ackEvents() resolves, with no other + // await before it. createFakeNative().settle() counts on that order to + // resolve after the row cache dropped the row; an await added above + // breaks that test helper (requests.test.ts "drops a completed row"). + try { + onAcked?.(event); + } catch (e) { + warn( + `delivery: updating the request rows after the ack of ${eventId} threw`, + e, + ); } }; @@ -247,7 +269,7 @@ export const createDelivery = ({ // A cancelled outcome has no handler, so it acks whether or not the key // is still defined. if (event.kind === 'cancelled') { - await ack(event.eventId); + await ack(event); return; } const definition = lookup(event.key); @@ -259,7 +281,7 @@ export const createDelivery = ({ return; } if (await runHandler(event, definition)) { - await ack(event.eventId); + await ack(event); } }; diff --git a/src/index.ts b/src/index.ts index 9abc1e55..d346c85c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -8,6 +8,12 @@ import NativeRNFileUploader from './NativeRNFileUploader'; import type { Spec } from './NativeRNFileUploader'; import { chunkPlan } from './chunkPlan'; import { createDelivery } from './delivery'; +import { + createRequestCache, + createSelector, + matches, + useRequestRows, +} from './requests'; import { createRegistry, DEFAULT_ENQUEUE_TIMEOUT_MS, @@ -20,11 +26,17 @@ import { import type { AddListener, ConfigureOptions, + Define, + Defined, + DefinedRequestsFilter, + Definition, Diagnostic, PauseScope, RequestRow, + RequestsFilter, StateEvent, UploadClient, + Vars, } from './types'; export * from './types'; @@ -131,9 +143,11 @@ export const createUploadClient = ({ native, lookup: (key) => definitions.get(key), emitState, + // Native forgets a completed or cancelled row at its ack, with no event. + onAcked: (event) => requests.acked(event), warn, }); - const { define } = createRegistry({ + const registry = createRegistry({ native, definitions, getSettings: () => settings, @@ -222,7 +236,16 @@ export const createUploadClient = ({ * when it has no entry file: a v9 upload's bytes, or a chunked file kept * after an enqueue that rejected E_STORAGE. */ - const cancel = (id: string): Promise => native.cancel(id); + const cancel = (id: string): Promise => + native.cancel(id).then(() => { + // Native forgets a settled entry with no event. Display only: the + // cancel has landed, so a throw here must not reject it. + try { + requests.cancelled(id); + } catch (e) { + warn(`cancel: updating the request rows after cancel(${id}) threw`, e); + } + }); /** * Persisted natively. Applies to queued and future entries whose @@ -313,6 +336,80 @@ export const createUploadClient = ({ } }) as AddListener; + const requests = createRequestCache({ + readNative: () => getRequests(), + addListener, + warn, + }); + const visibleRows = createSelector(requests.read, {}); + + const getCachedRequests = (filter?: RequestsFilter): readonly RequestRow[] => { + if (filter?.key === undefined && filter?.id === undefined) { + return filter?.includeCancelled === true + ? requests.read() + : visibleRows(); + } + return requests.read().filter((row) => matches(row, filter)); + }; + + const addRequestsListener = ( + listener: (rows: readonly RequestRow[]) => void, + ): { remove: () => void } => { + // No call while it registers: the caller reads getCachedRequests() after + // this returns. After that, a call only when the rows it gets changed, + // not when only a hidden cancelled row did. + let registering = true; + let last: readonly RequestRow[] | undefined; + const remove = requests.subscribe(() => { + if (registering) { + return; + } + const rows = visibleRows(); + if (rows === last) { + return; + } + last = rows; + try { + listener(rows); + } catch (e) { + warn('addRequestsListener: a listener threw', e); + } + }); + registering = false; + last = visibleRows(); + return { remove }; + }; + + const useRequests = (filter?: RequestsFilter): readonly RequestRow[] => + useRequestRows(requests.subscribe, requests.rows, filter); + + const useRequestsWithProgress = ( + filter?: RequestsFilter, + ): readonly RequestRow[] => + useRequestRows( + requests.subscribeWithProgress, + requests.rowsWithProgress, + filter, + ); + + // define() from the registry, plus the definition's own row hooks. The + // rows of a key are the rows its mutate() enqueued, so their vars are + // typed as the definition's vars. + const define = (( + definition: Definition, + ): Defined => { + const defined = registry.define(definition); + const { key } = defined; + type Rows = ReturnType['useRequests']>; + return { + ...defined, + useRequests: (filter?: DefinedRequestsFilter) => + useRequests({ ...filter, key }) as Rows, + useRequestsWithProgress: (filter?: DefinedRequestsFilter) => + useRequestsWithProgress({ ...filter, key }) as Rows, + }; + }) as Define; + const android = { /** * Fires when the Android upload notification is pressed. It never fires @@ -331,6 +428,10 @@ export const createUploadClient = ({ setWifiOnly, updateHeaders, getRequests, + useRequests, + useRequestsWithProgress, + addRequestsListener, + getCachedRequests, takeDiagnostics, addListener, chunkPlan, diff --git a/src/registry.ts b/src/registry.ts index a2bc0c4e..c956ac64 100644 --- a/src/registry.ts +++ b/src/registry.ts @@ -1,6 +1,5 @@ import type { Spec } from './NativeRNFileUploader'; import type { - Define, Defined, Definition, FormPart, @@ -120,8 +119,19 @@ type RegistryDeps = { now?: () => number; }; +/** + * What the registry's define() builds: the key and mutate(). The client adds + * the row hooks to make a full `Defined`. + */ +export type RegistryDefined = Pick< + Defined, + 'key' | 'mutate' +>; + export type Registry = { - define: Define; + define: ( + definition: Definition, + ) => RegistryDefined; }; declare const __DEV__: boolean | undefined; @@ -579,10 +589,11 @@ export const createRegistry = ({ now = Date.now, }: RegistryDeps): Registry => { // The overloads on Define keep the with-parser and without-parser shapes - // apart for callers. One implementation serves both. - const define = (( + // apart for callers. The client applies them; one implementation serves + // both. + const define = ( definition: Definition, - ): Defined => { + ): RegistryDefined => { const { key } = definition; if (typeof key !== 'string' || key.length === 0) { throw new Error('define: key must be a non-empty string'); @@ -653,8 +664,8 @@ export const createRegistry = ({ return { id: entry.id }; }; - return { key, mutate } as Defined; - }) as Define; + return { key, mutate } as RegistryDefined; + }; return { define }; }; diff --git a/src/requests.ts b/src/requests.ts new file mode 100644 index 00000000..79477070 --- /dev/null +++ b/src/requests.ts @@ -0,0 +1,706 @@ +/** + * The client's in-memory copy of the native queue rows, for display. It is + * read from native when the first listener arrives and patched by each + * `state` event, which carries the full row. A second view overlays the bytes + * of `progress` events. The library closes the gaps where native changes a + * row with no event, because its own JS makes those calls: the ack of a + * completed or cancelled outcome, and cancel() on a settled entry. + * + * Events are asynchronous, so a state event can arrive after the cache + * removed its row: Android emits a run's outcome before its last state event, + * and the ack of that outcome can forget the row in between. The ids the + * cache removed this way are kept, and their state events are checked + * against native instead of being applied. + * + * Nothing here sends, acknowledges, or cancels anything. A failure here can + * only make a row show late or stay longer. + */ +import { useMemo, useSyncExternalStore } from 'react'; +import type { + AddListener, + ProgressEvent, + RequestRow, + RequestsFilter, + RequestState, + StateEvent, +} from './types'; + +type Listener = () => void; + +const LIVE_STATES: readonly RequestState[] = [ + 'queued', + 'running', + 'awaiting-auth', + 'paused', +]; + +/** + * The most ids the cache keeps as forgotten. Ids acked earlier have no state + * event on the way, so only the ids acked in the last moments matter, and a + * burst of a few hundred acks at launch does not need a larger bound. An id + * pushed out is still covered for settled states: see onState(). + */ +export const MAX_FORGOTTEN = 256; + +/** + * Deep equality for JSON values, as `vars` and rows hold them. A field set to + * `undefined` counts as absent. It allocates nothing, because it runs on + * every state event and on every row of a native read. + */ +export const jsonEqual = (a: unknown, b: unknown): boolean => { + if (a === b) { + return true; + } + if ( + typeof a !== 'object' || + typeof b !== 'object' || + a === null || + b === null + ) { + return false; + } + const isArray = Array.isArray(a); + if (isArray !== Array.isArray(b)) { + return false; + } + if (isArray) { + const left = a as unknown[]; + const right = b as unknown[]; + if (left.length !== right.length) { + return false; + } + for (let i = 0; i < left.length; i++) { + if (!jsonEqual(left[i], right[i])) { + return false; + } + } + return true; + } + const left = a as Record; + const right = b as Record; + // Count the defined fields of each side; equal counts plus every left + // field equal means the same set of fields. + let fields = 0; + for (const k in left) { + if (left[k] === undefined) { + continue; + } + fields++; + if (!jsonEqual(left[k], right[k])) { + return false; + } + } + for (const k in right) { + if (right[k] !== undefined) { + fields--; + } + } + return fields === 0; +}; + +/** + * Row equality. The scalar fields that change on most transitions are + * compared first, so a changed row is found without walking its `vars`. + * `vars` is always compared: a same-id mutate() can replace it. + */ +export const rowEqual = (a: RequestRow, b: RequestRow): boolean => + a === b || + (a.id === b.id && + a.state === b.state && + a.updatedAt === b.updatedAt && + a.attempts === b.attempts && + a.bytesSent === b.bytesSent && + a.totalBytes === b.totalBytes && + a.nextAttemptAt === b.nextAttemptAt && + a.key === b.key && + jsonEqual(a, b)); + +const sameItems = (a: readonly T[], b: readonly T[]): boolean => { + if (a.length !== b.length) { + return false; + } + for (let i = 0; i < a.length; i++) { + if (a[i] !== b[i]) { + return false; + } + } + return true; +}; + +export const matches = (row: RequestRow, filter: RequestsFilter): boolean => + (filter.includeCancelled === true || row.state !== 'cancelled') && + (filter.key === undefined || row.key === filter.key) && + (filter.id === undefined || row.id === filter.id); + +// One id index per rows array. The cache makes a new array for each change, +// so every id-filtered selector of that change shares one build. +const indexes = new WeakMap< + readonly RequestRow[], + ReadonlyMap +>(); + +/** The rows of `rows` by id, built once per array. */ +export const indexById = ( + rows: readonly RequestRow[], +): ReadonlyMap => { + let index = indexes.get(rows); + if (!index) { + const built = new Map(); + for (const row of rows) { + built.set(row.id, row); + } + indexes.set(rows, built); + index = built; + } + return index; +}; + +const EMPTY: readonly RequestRow[] = []; + +/** + * A read of `source()` narrowed by `filter`. It returns the same array while + * the rows it keeps are the same objects, so a row it leaves out can change + * without a new result. `useSyncExternalStore` then re-renders only when the + * filtered rows change. With an `id`, it looks the row up in the shared + * index and does not walk the rows. + */ +export const createSelector = ( + source: () => readonly RequestRow[], + filter: RequestsFilter, +): (() => readonly RequestRow[]) => { + let lastSource: readonly RequestRow[] | undefined; + let lastResult: readonly RequestRow[] | undefined; + if (filter.id !== undefined) { + const { id } = filter; + let lastRow: RequestRow | undefined; + return () => { + const rows = source(); + if (rows === lastSource && lastResult) { + return lastResult; + } + lastSource = rows; + const row = indexById(rows).get(id); + const kept = row && matches(row, filter) ? row : undefined; + if (!lastResult || kept !== lastRow) { + lastRow = kept; + lastResult = kept ? [kept] : EMPTY; + } + return lastResult; + }; + } + return () => { + const rows = source(); + if (rows === lastSource && lastResult) { + return lastResult; + } + const next = rows.filter((row) => matches(row, filter)); + lastSource = rows; + if (!lastResult || !sameItems(next, lastResult)) { + lastResult = next; + } + return lastResult; + }; +}; + +// One selector per source and filter without an id, shared by every hook +// that asks for it, so all unfiltered or same-key hooks filter the rows once +// per change. The number of entries is the number of distinct keys used. +const shared = new WeakMap< + () => readonly RequestRow[], + Map readonly RequestRow[]> +>(); + +/** `createSelector()` for a filter without an id, shared per source. */ +export const sharedSelector = ( + source: () => readonly RequestRow[], + filter: Omit, +): (() => readonly RequestRow[]) => { + let bySource = shared.get(source); + if (!bySource) { + bySource = new Map(); + shared.set(source, bySource); + } + const includeCancelled = filter.includeCancelled === true; + const name = `${includeCancelled ? 1 : 0}:${filter.key ?? ''}:${ + filter.key === undefined ? 0 : 1 + }`; + let selector = bySource.get(name); + if (!selector) { + selector = createSelector(source, { key: filter.key, includeCancelled }); + bySource.set(name, selector); + } + return selector; +}; + +type RequestCacheDeps = { + /** The client's getRequests() without a filter. May throw. */ + readNative: () => RequestRow[]; + /** The client's addListener(), so unhandled-key rows arrive too. */ + addListener: AddListener; + warn: (message: string, detail?: unknown) => void; +}; + +export type RequestCache = { + /** Every cached row, cancelled ones included. Stable until a row changes. */ + rows: () => readonly RequestRow[]; + /** `rows()` with the bytes of the progress events since each row's last state. */ + rowsWithProgress: () => readonly RequestRow[]; + /** Called when a row appears, changes, or goes. Returns the unsubscribe. */ + subscribe: (listener: Listener) => () => void; + /** Also called on every progress event for a cached row. */ + subscribeWithProgress: (listener: Listener) => () => void; + /** + * For reads outside React: with no listener, reads native first, so the + * rows are never out of date. + */ + read: () => readonly RequestRow[]; + /** The ack of an outcome of `id` resolved. */ + acked: (outcome: { id: string; kind: string }) => void; + /** cancel(id) resolved. */ + cancelled: (id: string) => void; +}; + +export const createRequestCache = ({ + readNative, + addListener, + warn, +}: RequestCacheDeps): RequestCache => { + // Undefined until the first read. + let rows: readonly RequestRow[] | undefined; + let withProgress: readonly RequestRow[] | undefined; + // The bytes of the last progress event of each row, and the overlaid row + // built from them, so an unchanged row keeps its identity. + const progress = new Map(); + const overlaid = new Map< + string, + { base: RequestRow; event: ProgressEvent; row: RequestRow } + >(); + const rowListeners = new Set(); + const progressListeners = new Set(); + let stateSubscription: { remove: () => void } | undefined; + // Native progress is listened to only while a progress view is mounted. + let progressSubscription: { remove: () => void } | undefined; + // Ids whose row native forgot, or may have forgotten, with no event. A + // state event for one of them can be late, so it is checked against + // native. Oldest first; an id leaves when native shows a row for it again. + // This relies on native events reaching JS in the order native sent them: + // a read can show a same-id mutate()'s new row while the old life's last + // event is still on the way, and that event then replaces the row until + // the new life's own next event corrects it. + const forgotten = new Set(); + + const notify = (listeners: Set): void => { + listeners.forEach((listener) => listener()); + }; + + const markForgotten = (id: string): void => { + // Delete first, so a repeated id moves to the newest place. + forgotten.delete(id); + forgotten.add(id); + if (forgotten.size > MAX_FORGOTTEN) { + const oldest = forgotten.values().next(); + if (!oldest.done) { + forgotten.delete(oldest.value); + } + } + }; + + /** After a read of every native row: an id native shows is not forgotten. */ + const unforgetShown = (current: readonly RequestRow[]): void => { + if (forgotten.size === 0) { + return; + } + for (const row of current) { + forgotten.delete(row.id); + } + }; + + /** True when the row had progress to forget. */ + const forgetProgress = (id: string): boolean => { + overlaid.delete(id); + return progress.delete(id); + }; + + const commit = (next: readonly RequestRow[]): void => { + if (next === rows) { + return; + } + rows = next; + withProgress = undefined; + // Progress of a row that left is not needed any more. + if (progress.size > 0) { + const index = indexById(next); + progress.forEach((_event, id) => { + if (!index.has(id)) { + forgetProgress(id); + } + }); + } + notify(rowListeners); + notify(progressListeners); + }; + + /** Native's rows, or undefined when the read throws. */ + const readSafely = (): RequestRow[] | undefined => { + try { + const current = readNative(); + unforgetShown(current); + return current; + } catch (e) { + warn( + 'requests: getRequests failed, so the cached rows were not checked against native', + e, + ); + return undefined; + } + }; + + /** + * Native's rows, with the cached object kept for each row that did not + * change, and the cached array kept when no row changed. + */ + const reconcile = ( + cachedRows: readonly RequestRow[], + current: RequestRow[], + ): readonly RequestRow[] => { + const cached = indexById(cachedRows); + const next = current.map((row) => { + const old = cached.get(row.id); + return old && rowEqual(old, row) ? old : row; + }); + return sameItems(next, cachedRows) ? cachedRows : next; + }; + + /** Checks every row against native. Keeps the rows when the read fails. */ + const refresh = (): void => { + const current = readSafely(); + if (!rows) { + // The first read: nobody has seen rows yet, so nobody is notified. + rows = current ?? []; + return; + } + if (current) { + commit(reconcile(rows, current)); + } + }; + + const ensureRows = (): readonly RequestRow[] => { + if (!rows) { + refresh(); + } + return rows ?? []; + }; + + const remove = (id: string): void => { + forgetProgress(id); + if (rows && indexById(rows).has(id)) { + commit(rows.filter((row) => row.id !== id)); + } + }; + + /** Removes a row that native forgot, and keeps its id for late events. */ + const forget = (id: string): void => { + markForgotten(id); + remove(id); + }; + + /** Puts `row` in the cache, in place of the row with its id. */ + const upsert = (row: RequestRow): void => { + const hadProgress = forgetProgress(row.id); + if (!rows) { + return; + } + const cached = indexById(rows).get(row.id); + if (!cached) { + commit([...rows, row]); + return; + } + if (rowEqual(cached, row)) { + // The row is the same, but the progress view showed newer bytes. + if (hadProgress) { + withProgress = undefined; + notify(progressListeners); + } + return; + } + commit(rows.map((other) => (other === cached ? row : other))); + }; + + /** Asks native for one row: replaces it, or drops it when native has none. */ + /** False when the native read failed and nothing changed. */ + const refreshId = (id: string): boolean => { + const current = readSafely(); + if (!current) { + return false; + } + const row = current.find((candidate) => candidate.id === id); + if (row) { + upsert(row); + } else { + forget(id); + } + return true; + }; + + // A state event carries the full row, so it replaces the cached one. Its + // bytes are the row's own, so the progress seen before it is out of date. + const onState = (event: StateEvent): void => { + if (!rows) { + return; + } + if (event.reason === 'unhandled-key') { + // JS builds this row from the journaled outcome, not from the entry. + // A same-id mutate() may have reopened the entry since, so ask native. + refreshId(event.id); + return; + } + const held = indexById(rows).has(event.id); + // A forgotten id: the event may be older than the forget, or be the + // first event of a same-id mutate(). A settled state for an id the cache + // does not hold: a new life starts live, so this is a late event too, + // for example of an id pushed out of `forgotten`. Ask native. + if ( + forgotten.has(event.id) || + (!held && !LIVE_STATES.includes(event.state)) + ) { + if (refreshId(event.id)) { + return; + } + // The read failed. A live state, or a row already shown, is most + // likely the new life of a reused id: show it rather than hide that + // life until the next read. + if (!held && !LIVE_STATES.includes(event.state)) { + return; + } + } + upsert(event); + }; + + const onProgress = (event: ProgressEvent): void => { + if (!rows || !indexById(rows).has(event.id)) { + return; + } + progress.set(event.id, event); + withProgress = undefined; + notify(progressListeners); + }; + + const listening = (): boolean => + rowListeners.size + progressListeners.size > 0; + + // These run inside native's event callbacks. A throw there would surface + // as an uncaught error and stop the app's later listeners of that event. + const guarded = + (name: string, handle: (event: E) => void) => + (event: E): void => { + try { + handle(event); + } catch (e) { + warn(`requests: updating the rows from a ${name} event threw`, e); + } + }; + const onStateGuarded = guarded('state', onState); + const onProgressGuarded = guarded('progress', onProgress); + + const start = (): void => { + stateSubscription = addListener('state', onStateGuarded); + // Always check: an event emitted between an earlier read (a render) and + // this subscribe went to no listener. reconcile() keeps the same array + // when nothing changed, so this adds no render. + refresh(); + }; + + // The rows stay, so a component that replaces the last listener in one + // commit does not render twice. The next start() checks them against native. + const stop = (): void => { + stateSubscription?.remove(); + stateSubscription = undefined; + }; + + const startProgress = (): void => { + progressSubscription = addListener('progress', onProgressGuarded); + }; + + const stopProgress = (): void => { + progressSubscription?.remove(); + progressSubscription = undefined; + progress.clear(); + overlaid.clear(); + withProgress = undefined; + }; + + const subscriber = + (listeners: Set) => + (listener: Listener): (() => void) => { + const first = !listening(); + const firstProgress = + listeners === progressListeners && listeners.size === 0; + // A wrapper, so the same function subscribed twice is two entries. + const entry = () => listener(); + listeners.add(entry); + if (firstProgress) { + startProgress(); + } + if (first) { + start(); + } + return () => { + if (!listeners.delete(entry)) { + return; + } + if (listeners === progressListeners && listeners.size === 0) { + stopProgress(); + } + if (!listening()) { + stop(); + } + }; + }; + + const rowsWithProgress = (): readonly RequestRow[] => { + if (withProgress) { + return withProgress; + } + const base = ensureRows(); + if (progress.size === 0) { + withProgress = base; + return base; + } + const next = base.map((row) => { + const event = progress.get(row.id); + if (!event) { + return row; + } + const memo = overlaid.get(row.id); + if (memo && memo.base === row && memo.event === event) { + return memo.row; + } + const merged = { + ...row, + bytesSent: event.bytesSent, + totalBytes: event.totalBytes, + }; + overlaid.set(row.id, { base: row, event, row: merged }); + return merged; + }); + withProgress = sameItems(next, base) ? base : next; + return withProgress; + }; + + /** + * With nothing listening, the rows may already be out of date and nobody + * shows them. Drop them instead of patching them: the next subscribe or + * read reads native. This keeps native reads out of the delivery lane + * while the journal drains at launch. + */ + const dropWhenIdle = (): boolean => { + if (listening()) { + return false; + } + rows = undefined; + withProgress = undefined; + return true; + }; + + const acked = ({ id, kind }: { id: string; kind: string }): void => { + // Native forgets a row at the ack of its completed or cancelled outcome. + const forgets = kind === 'completed' || kind === 'cancelled'; + if (dropWhenIdle()) { + // A state event queued before the next subscribe still reaches it. + if (forgets) { + markForgotten(id); + } + return; + } + const row = rows ? indexById(rows).get(id) : undefined; + if (kind === 'error') { + // Native keeps an error row. iOS can move a live row to its outcome + // during the ack with no event, so a row not shown as an error is + // checked. + if (row && row.state !== 'error') { + refreshId(id); + } + return; + } + if (!forgets) { + return; + } + if (!row || row.state === kind) { + forget(id); + return; + } + // The row moved on, for example a cancelled entry that a same-id + // mutate() reopened, whose old ack does not forget it; or a live row that + // native settled and forgot during the ack, whose last state event is + // still on its way. Ask native. + refreshId(id); + }; + + const cancelled = (id: string): void => { + if (dropWhenIdle()) { + markForgotten(id); + return; + } + // Native forgets a completed or error entry at once, with no event. It + // moves a live entry to 'cancelled', usually with the event already + // here, and keeps it until the ack. So a 'cancelled' row may be one + // native keeps or one this cancel forgot (an entry cancelled before). A + // row still shown live may already be settled in native, with its state + // event on the way, and this cancel forgot it. Ask native for both. + const row = rows ? indexById(rows).get(id) : undefined; + if (!row) { + markForgotten(id); + return; + } + if (row.state === 'cancelled' || LIVE_STATES.includes(row.state)) { + refreshId(id); + } else { + forget(id); + } + }; + + return { + rows: ensureRows, + rowsWithProgress, + subscribe: subscriber(rowListeners), + subscribeWithProgress: subscriber(progressListeners), + read: () => { + if (listening()) { + return rows ?? []; + } + // Nothing listens, so nothing holds these rows: take native's as they + // are. The next start() reconciles them. + const current = readSafely(); + if (current) { + rows = current; + withProgress = undefined; + } + return rows ?? []; + }, + acked, + cancelled, + }; +}; + +/** + * The hook behind `useRequests()` and `useRequestsWithProgress()`. The + * filter's fields, not the object, key the selector, so an inline filter + * object does not make a new result each render. + */ +export const useRequestRows = ( + subscribe: (listener: Listener) => () => void, + source: () => readonly RequestRow[], + filter: RequestsFilter | undefined, +): readonly RequestRow[] => { + const key = filter?.key; + const id = filter?.id; + const includeCancelled = filter?.includeCancelled === true; + const getSnapshot = useMemo( + () => + id === undefined + ? sharedSelector(source, { key, includeCancelled }) + : createSelector(source, { key, id, includeCancelled }), + [source, key, id, includeCancelled], + ); + return useSyncExternalStore(subscribe, getSnapshot); +}; diff --git a/src/testing.ts b/src/testing.ts index 1b97d130..4aef5f59 100644 --- a/src/testing.ts +++ b/src/testing.ts @@ -144,7 +144,8 @@ export type FakeNative = Spec & { /** * Journals the outcome of entry `id`, moves its row to the outcome state, * and emits it on onSettled when a listener exists. Resolves with the event - * after delivery acks it. Rejects after `ackTimeoutMs` when no ack comes: + * after delivery acks it and the client has handled the ack, so the row + * hooks no longer show a forgotten row. Rejects after `ackTimeoutMs` when no ack comes: * the handler rejected, the key has no definition, or configure() was not * called. To test those cases, create the fake with a short * `ackTimeoutMs` and expect settle() to reject. @@ -154,7 +155,11 @@ export type FakeNative = Spec & { outcome: FakeOutcome, overrides?: SettledOverrides, ) => Promise; - /** Adds or replaces rows, for example legacy rows that native imported. */ + /** + * Adds or replaces rows, for example legacy rows that native imported. It + * emits no event, so a mounted row hook sees the rows only after it mounts + * again. + */ seedRows: (rows: RequestRow[]) => void; /** * Adds outcomes to the journal. getUnacknowledgedEvents() returns them, so @@ -587,8 +592,8 @@ export const createFakeNative = ( }), ), - ackEvents: (ids) => - run('ackEvents', () => { + ackEvents: (ids) => { + const acked = run('ackEvents', () => { ids.forEach((eventId) => { ackedEventIds.push(eventId); const index = journal.findIndex((e) => e.eventId === eventId); @@ -608,15 +613,29 @@ export const createFakeNative = ( rows.delete(event.id); } }); - // Waiters resolve after the journal and rows are up to date. - ids.forEach((eventId) => { - const waiter = ackWaiters.get(eventId); - if (waiter) { - ackWaiters.delete(eventId); - waiter(); - } - }); - }), + }); + // Waiters resolve after the journal and rows are up to date, and one + // promise turn after the caller's own continuation on this promise. + // So when settle() resolves, the client has handled the ack too, for + // example dropped the row from its row cache. This depends on + // delivery's ack() handling the ack in the turn after ackEvents() + // resolves (see the comment there). A macrotask would not depend on it, + // but it would make settle() wait on timers, which Jest fake timers + // hold. + void acked + .then(() => undefined) + .then(() => { + ids.forEach((eventId) => { + const waiter = ackWaiters.get(eventId); + if (waiter) { + ackWaiters.delete(eventId); + waiter(); + } + }); + }) + .catch(() => undefined); + return acked; + }, takeDiagnostics: () => run('takeDiagnostics', () => diff --git a/src/types.ts b/src/types.ts index 9fcc2035..952bf9f5 100644 --- a/src/types.ts +++ b/src/types.ts @@ -229,6 +229,31 @@ export type RequestRow = { nextAttemptAt?: number; }; +/** + * Which cached rows `useRequests()`, `useRequestsWithProgress()` and + * `getCachedRequests()` return. `key` and `id` match as in `getRequests()`. + * Cancelled rows are left out unless `includeCancelled` is true: such a row + * has no work left, and native forgets it at its ack. + */ +export type RequestsFilter = { + key?: string; + id?: string; + includeCancelled?: boolean; +}; + +/** A definition's rows: the filter has no `key`, because it is the definition's. */ +export type DefinedRequestsFilter = Omit; + +/** + * A row of one definition, with `vars` typed as that definition's vars. The + * type is a promise the library cannot check: `vars` is what native stored + * from `JSON.stringify`, so a `Date` reads back as a string, and an entry + * enqueued under an older vars type keeps its old shape. + */ +export type DefinedRequestRow = Omit & { + vars: V; +}; + /** * A `state` event. `reason: 'unhandled-key'` reports an outcome whose key has * no definition. The row carries the entry's real state, and the library keeps @@ -316,6 +341,14 @@ export type Defined = { mutate: [V] extends [null] ? (vars?: null, options?: { id?: string }) => Promise<{ id: string }> : (vars: V, options?: { id?: string }) => Promise<{ id: string }>; + /** The client's `useRequests()` for this definition's key, with typed vars. */ + useRequests: ( + filter?: DefinedRequestsFilter, + ) => readonly DefinedRequestRow[]; + /** The client's `useRequestsWithProgress()` for this definition's key. */ + useRequestsWithProgress: ( + filter?: DefinedRequestsFilter, + ) => readonly DefinedRequestRow[]; }; /** @@ -466,6 +499,34 @@ export type UploadClient = { setWifiOnly: (enabled: boolean) => Promise; updateHeaders: (patch: Record) => Promise; getRequests: (filter?: { key?: string; id?: string }) => RequestRow[]; + /** + * React hook. The cached rows: read from native when the first listener + * arrives, then patched by events. Re-renders when a row appears, changes + * state, or goes, not on progress. The same array returns while nothing it + * holds changed. + */ + useRequests: (filter?: RequestsFilter) => readonly RequestRow[]; + /** + * React hook. `useRequests()` with the bytes of each progress event, so it + * also re-renders on progress. Use it only where bytes are shown. + */ + useRequestsWithProgress: (filter?: RequestsFilter) => readonly RequestRow[]; + /** + * Calls `listener` with `getCachedRequests()` each time those rows change: + * a row appears, changes state, or goes. There is no call at registration; + * read `getCachedRequests()` after this returns. Keeps the cache up to date + * while subscribed. + */ + addRequestsListener: (listener: (rows: readonly RequestRow[]) => void) => { + remove: () => void; + }; + /** + * The rows the hooks see, for code outside React. With no listener and no + * mounted hook, it reads native and returns a new array each call. A result + * filtered by `key` or `id` is a new array each call, so it must not be a + * `useSyncExternalStore` snapshot. + */ + getCachedRequests: (filter?: RequestsFilter) => readonly RequestRow[]; takeDiagnostics: () => Promise; addListener: AddListener; chunkPlan: (