From 3aa972e8b8b53a3e486c3c8f2b5a3ae600fdffa4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Johannes=20K=C3=B6hler?= Date: Sat, 31 Oct 2026 08:50:40 +0100 Subject: [PATCH] Add Collection.observeRecord for records that may be missing Adds Collection.observeRecord(id), a failsafe version of findAndObserve(id). It emits null while the record is missing, the record once it exists or is created, and null again instead of completing when it is deleted, and keeps observing throughout. Relations get a failsafe option that makes observe() use observeRecord, and @experimentalFailsafe sets it, so failsafe relations recover when a missing related record is created. Model and Collection now notify a copy of their subscriber lists, so subscribers can unsubscribe while being notified. In TypeScript, Relation accepts a nullable record type. Refs Nozbe/WatermelonDB#1257 and Nozbe/WatermelonDB#1719. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG-Unreleased.md | 2 + docs-website/docs/docs/CRUD.md | 3 +- docs-website/docs/docs/Components.md | 7 + docs-website/docs/docs/Relation.md | 13 + src/Collection/index.d.ts | 8 + src/Collection/index.js | 147 +++++++- src/Collection/test.js | 357 ++++++++++++++++++ src/Model/index.js | 6 +- src/Relation/helpers.d.ts | 4 +- src/Relation/helpers.js | 26 +- src/Relation/index.d.ts | 16 +- src/Relation/index.js | 6 + src/Relation/test.js | 118 ++++++ src/decorators/experimentalFailsafe/index.js | 20 +- src/decorators/experimentalFailsafe/test.js | 157 ++++++++ .../withObservables/withObservables.test.js | 51 +++ 16 files changed, 916 insertions(+), 25 deletions(-) create mode 100644 src/decorators/experimentalFailsafe/test.js diff --git a/CHANGELOG-Unreleased.md b/CHANGELOG-Unreleased.md index 9d43e2bdd..5f71752ab 100644 --- a/CHANGELOG-Unreleased.md +++ b/CHANGELOG-Unreleased.md @@ -6,6 +6,8 @@ ### New features +- New `Collection.observeRecord(id)`: a failsafe version of `findAndObserve(id)` that emits `null` while the record does not exist (instead of erroring), emits the record once it is created, and emits `null` (instead of completing) when it is deleted. `@experimentalFailsafe` relations now use it, so their `observe()` recovers when a missing related record is created later + ### Fixes - [LokiJS] Multitab sync issue fix diff --git a/docs-website/docs/docs/CRUD.md b/docs-website/docs/docs/CRUD.md index fc443f211..af5279f97 100644 --- a/docs-website/docs/docs/CRUD.md +++ b/docs-website/docs/docs/CRUD.md @@ -111,7 +111,8 @@ await somePost.destroyPermanently() // permanent - `Model.observe()` - usually you only use this [when connecting records to components](./Components.md), but you can manually observe a record outside of React components. The returned [RxJS](https://github.com/reactivex/rxjs) `Observable` will emit the record immediately upon subscription, and then every time the record is updated. If the record is deleted, the Observable will complete. - `Query.observe()`, `Relation.observe()` — analagous to the above, but for [Queries](./Query.md) and [Relations](./Relation.md) - `Query.observeWithColumns()` - used for [sorted lists](./Components.md) -- `Collection.findAndObserve(id)` — same as using `.find(id)` and then calling `record.observe()` +- `Collection.findAndObserve(id)` — same as using `.find(id)` and then calling `record.observe()`. The Observable errors if the record does not exist, and completes if it's deleted. +- `Collection.observeRecord(id)` — failsafe version of `findAndObserve(id)` for records that might not exist (yet). Emits `null` while the record is missing, emits the record once it exists or is created later (e.g. arrives via sync), and emits `null` again (instead of completing) when it's deleted. Use it when rendering a missing record as a placeholder is better than an error. - `Model.prepareUpdate()`, `Collection.prepareCreate`, `Database.batch` — used for [batch updates](./Writers.md) - `Database.unsafeResetDatabase()` destroys the whole database - [be sure to see this comment before using it](https://github.com/Nozbe/WatermelonDB/blob/22188ee5b6e3af08e48e8af52d14e0d90db72925/src/Database/index.js#L131) - To override the `record.id` during the creation, e.g. to sync with a remote database, you can do it by `record._raw` property. Be aware that the `id` must be of type `string`. diff --git a/docs-website/docs/docs/Components.md b/docs-website/docs/docs/Components.md index ceb9cd98f..bb59e361d 100644 --- a/docs-website/docs/docs/Components.md +++ b/docs-website/docs/docs/Components.md @@ -161,6 +161,13 @@ withObservables(['post'], ({ post }) => ({ post: database.get('posts').findAndObserve(postId) })) ``` + If the post does not exist, `findAndObserve` errors, and the error is re-thrown when rendering (use an Error Boundary to catch it). If the record might legitimately be missing — for example, it hasn't been synced yet, or it was deleted — use **observeRecord** instead: + ```js + withObservables(['postId'], ({ postId, database }) => ({ + post: database.get('posts').observeRecord(postId) + })) + ``` + Your component then receives `post={null}` while the post is missing (render a placeholder), and is re-rendered with the post as soon as it's created. If the post is deleted, `post` becomes `null` again. Similarly, mark a relation with `@experimentalFailsafe` to make `Relation.observe()` emit a fallback value (instead of erroring) while the related record is missing, and recover once it's created. 1. **RxJS transformations**. The values returned by `Model.observe()`, `Query.observe()`, `Relation.observe()` are [RxJS Observables](https://github.com/ReactiveX/rxjs). You can use standard transforms like mapping, filtering, throttling, startWith to change when and how the component is re-rendered. 1. **Custom Observables**. `withObservables` is a general-purpose HOC for Observables, not just Watermelon. You can create new props from any `Observable`. diff --git a/docs-website/docs/docs/Relation.md b/docs-website/docs/docs/Relation.md index acddddda4..be0b0a041 100644 --- a/docs-website/docs/docs/Relation.md +++ b/docs-website/docs/docs/Relation.md @@ -63,6 +63,19 @@ withObservables(['comment'], ({ comment }) => ({ The component will now have an `author` prop containing a `User`, and will re-render both when the user changes (e.g. comment's author changes its name), but also when a new author is assigned to the comment (if that was possible). +If the related record might not exist (e.g. it hasn't been synced yet), observing the relation errors. To render a placeholder instead, mark the relation with `@experimentalFailsafe`. `author` will then be `undefined` (or the fallback value you pass to the decorator) while the user is missing, and the component re-renders with the `User` once it's created: + +```js +import { experimentalFailsafe } from '@nozbe/watermelondb/decorators' + +class Comment extends Model { + // ... + @experimentalFailsafe() @relation('users', 'author_id') author +} +``` + +If you use TypeScript or Flow, declare such a relation with a nullable type (e.g. `author: Relation`), since it can emit `undefined` (or your fallback). The same goes for relations whose ID is optional, which emit `null` when the ID is not set. + ### Fetching To simply get the related record, use `fetch`. You might need it [in a Writer](./Writers.md) diff --git a/src/Collection/index.d.ts b/src/Collection/index.d.ts index af47bb402..5595a0162 100644 --- a/src/Collection/index.d.ts +++ b/src/Collection/index.d.ts @@ -41,6 +41,11 @@ export default class Collection { // (with the same semantics as when calling `model.observe()`) findAndObserve(id: RecordId): Observable + // Observes the record with the given ID, treating a missing record as `null`. + // Failsafe version of `findAndObserve`: emits `null` while the record does not exist, + // the record once it exists (or is created later), and `null` again when it is deleted + observeRecord(id: RecordId): Observable + // Query records of this type query: ArrayOrSpreadFn> @@ -87,6 +92,9 @@ export default class Collection { // Fetches exactly one record (See: Collection.find) _fetchRecord(id: RecordId, callback: ResultCallback): void + // Fetches exactly one record, or `null` if it does not exist (See: Collection.observeRecord) + _fetchRecordIfExists(id: RecordId, callback: ResultCallback): void + _applyChangesToCache(operations: CollectionChangeSet): void _notify(operations: CollectionChangeSet): void diff --git a/src/Collection/index.js b/src/Collection/index.js index 5b8405e51..cf6b9c075 100644 --- a/src/Collection/index.js +++ b/src/Collection/index.js @@ -116,6 +116,130 @@ export default class Collection { }) } + /** + * Observes the record with the given ID, treating a missing record as `null`. + * + * This is a failsafe version of `findAndObserve()`. The returned Observable never errors or + * completes because of the record's existence. Instead, it emits: + * - the record, if it exists, and then again every time it changes + * - `null`, if the record does not exist (yet) or is marked as deleted + * - the record, once a record with this ID is created (e.g. arrives via sync) + * - `null` again, if the record is deleted — and the record again, if it is re-created + * + * Use this instead of `findAndObserve()` when the record might not exist, e.g. when observing + * a related record that may arrive later, so that your UI can render a placeholder instead of + * throwing an error. + */ + observeRecord(id: RecordId): Observable { + return Observable.create((observer) => { + let unsubscribed = false + let observedRecord: ?Record = null + let unsubscribeFromRecord: ?Unsubscribe = null + let unsubscribeFromCollection: ?Unsubscribe = null + // Set once a change notification for this ID arrives - from then on, the record cache is + // the source of truth, and the (possibly outdated) result of the initial fetch is ignored + let initialFetchSuperseded = false + // Whether the last value emitted was `null` (so that we don't emit it twice in a row) + let isMissingEmitted = false + + const emitMissing = () => { + if (!isMissingEmitted) { + isMissingEmitted = true + observer.next(null) + } + } + + const stopObservingRecord = () => { + unsubscribeFromRecord && unsubscribeFromRecord() + unsubscribeFromRecord = null + observedRecord = null + } + + const stopObservingCollection = () => { + unsubscribeFromCollection && unsubscribeFromCollection() + unsubscribeFromCollection = null + } + + // Caches are updated before change notifications are sent, so at notification time the + // cache tells us if a (non-deleted) record with this ID currently exists + const getLiveCachedRecord = (): ?Record => { + const record = this._cache.get(id) + return record && record.syncStatus !== 'deleted' ? record : null + } + + const observeExistingRecord = (record: Record) => { + stopObservingCollection() + stopObservingRecord() + observedRecord = record + // Subscribe before emitting, so that if the consumer unsubscribes synchronously, + // the teardown can remove this listener + unsubscribeFromRecord = record.experimentalSubscribe((isDeleted) => { + if (unsubscribed) { + return + } + if (isDeleted) { + stopObservingRecord() + // The record might have been re-created in the same batch + const recreatedRecord = getLiveCachedRecord() + recreatedRecord ? observeExistingRecord(recreatedRecord) : observeMissingRecord() + } else if (this._cache.get(id) === record) { + // (With batched notifications, the record might have been deleted in the meantime - + // in that case, a deletion notification follows, so don't emit the deleted record) + observer.next(record) + } + }) + isMissingEmitted = false + observer.next(record) + } + + // While the record is missing, watch for a record with this ID being created. (Once it + // exists, we only listen to that record, so that we don't have to scan every change) + const startObservingCollection = () => { + unsubscribeFromCollection = this.experimentalSubscribe((changes) => { + if (unsubscribed || !changes.some(({ record }) => record.id === id)) { + return + } + initialFetchSuperseded = true + const record = getLiveCachedRecord() + // (If the record was created and deleted while the initial fetch is pending, we still + // need to emit the initial `null`) + record ? observeExistingRecord(record) : emitMissing() + }) + } + + const observeMissingRecord = () => { + startObservingCollection() + emitMissing() + } + + startObservingCollection() + // NOTE: A fetched raw record is always turned into a cached model (even if this fetch is + // superseded), because the adapter marks it as sent to JS - dropping it would leave the + // adapter-side and JS record caches out of sync. Adapters process calls in order, so a find + // result can't be older than a change notification that arrives before it. + this._fetchRecordIfExists(id, (result) => { + if (unsubscribed || initialFetchSuperseded || observedRecord) { + return + } + const record = result.value + if (result.error) { + observer.error(result.error) + } else if (record && record.syncStatus !== 'deleted') { + observeExistingRecord(record) + } else { + // Missing, or marked as deleted (adapters still find records marked as deleted) + emitMissing() + } + }) + + return () => { + unsubscribed = true + stopObservingCollection() + stopObservingRecord() + } + }) + } + /*:: query: ArrayOrSpreadFn> */ /** * Returns a `Query` with conditions given. @@ -217,6 +341,18 @@ export default class Collection { // Fetches exactly one record (See: Collection.find) _fetchRecord(id: RecordId, callback: ResultCallback): void { + this._fetchRecordIfExists(id, (result) => + callback( + mapValue((record) => { + invariant(record, `Record ${this.table}#${id} not found`) + return record + }, result), + ), + ) + } + + // Fetches exactly one record, or `null` if it does not exist (See: Collection.observeRecord) + _fetchRecordIfExists(id: RecordId, callback: ResultCallback): void { if (typeof id !== 'string') { callback({ error: new Error(`Invalid record ID ${this.table}#${id}`) }) return @@ -231,10 +367,10 @@ export default class Collection { this.database.adapter.underlyingAdapter.find(this.table, id, (result) => callback( - mapValue((rawRecord) => { - invariant(rawRecord, `Record ${this.table}#${id} not found`) - return this._cache.recordFromQueryResult(rawRecord) - }, result), + mapValue( + (rawRecord) => (rawRecord ? this._cache.recordFromQueryResult(rawRecord) : null), + result, + ), ), ) } @@ -257,7 +393,8 @@ export default class Collection { ]): void => { subscriber(operations) } - this._subscribers.forEach(collectionChangeNotifySubscribers) + // Iterate over a copy, so that subscribers can safely unsubscribe while being notified + this._subscribers.slice().forEach(collectionChangeNotifySubscribers) this.changes.next(operations) const collectionChangeNotifyModels = ({ record, type }: CollectionChange): void => { diff --git a/src/Collection/test.js b/src/Collection/test.js index 3607aab9b..54ff4c1b3 100644 --- a/src/Collection/test.js +++ b/src/Collection/test.js @@ -1,3 +1,4 @@ +import { filter, take } from 'rxjs/operators' import { expectToRejectWithMessage } from '../__tests__/utils' import { noop } from '../utils/fp' @@ -355,3 +356,359 @@ describe('Collection observation', () => { expect(subscriber).toHaveBeenCalledTimes(4) }) }) + +describe('findAndObserve', () => { + it('observes an existing record and completes when it is deleted', async () => { + const { database, tasks } = mockDatabase() + const t1 = await database.write(() => tasks.create()) + + const observer = jest.fn() + const complete = jest.fn() + tasks.findAndObserve(t1.id).subscribe(observer, undefined, complete) + + await new Promise(process.nextTick) + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(t1) + + await database.write(() => t1.update()) + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(t1) + + await database.write(() => t1.markAsDeleted()) + expect(observer).toHaveBeenCalledTimes(2) + expect(complete).toHaveBeenCalledTimes(1) + }) + it('errors if the record does not exist', async () => { + const { tasks } = mockDatabase() + + const observer = jest.fn() + const error = jest.fn() + tasks.findAndObserve('missing').subscribe(observer, error) + + await new Promise(process.nextTick) + expect(observer).toHaveBeenCalledTimes(0) + expect(error).toHaveBeenCalledTimes(1) + expect(error.mock.calls[0][0].message).toMatch('Record mock_tasks#missing not found') + }) +}) + +describe('observeRecord', () => { + const createWithId = (database, collection, id) => + database.write(() => + collection.create((record) => { + record._raw.id = id + }), + ) + + it('emits null if the record does not exist', async () => { + const { tasks } = mockDatabase() + + const observer = jest.fn() + const error = jest.fn() + const complete = jest.fn() + tasks.observeRecord('missing').subscribe(observer, error, complete) + + await new Promise(process.nextTick) + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + expect(error).toHaveBeenCalledTimes(0) + expect(complete).toHaveBeenCalledTimes(0) + }) + it('emits the record once a record with this id is created', async () => { + const { database, tasks } = mockDatabase() + + const observer = jest.fn() + const subscription = tasks.observeRecord('later').subscribe(observer) + + await new Promise(process.nextTick) + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + + // unrelated records don't matter + await database.write(() => tasks.create()) + expect(observer).toHaveBeenCalledTimes(1) + + const t1 = await createWithId(database, tasks, 'later') + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(t1) + + // ...and then keeps observing the record + await database.write(() => t1.update()) + expect(observer).toHaveBeenCalledTimes(3) + expect(observer).toHaveBeenLastCalledWith(t1) + + subscription.unsubscribe() + await database.write(() => t1.update()) + expect(observer).toHaveBeenCalledTimes(3) + }) + it('emits null instead of completing when the record is deleted, and recovers', async () => { + const { database, tasks } = mockDatabase() + const t1 = await createWithId(database, tasks, 't1') + + const observer = jest.fn() + const error = jest.fn() + const complete = jest.fn() + const subscription = tasks.observeRecord('t1').subscribe(observer, error, complete) + + // record is cached, so it's emitted synchronously + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(t1) + + await database.write(() => t1.markAsDeleted()) + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(null) + expect(complete).toHaveBeenCalledTimes(0) + + // purge the deleted record (like sync does), then re-create a record with the same id + await database.adapter.destroyDeletedRecords('mock_tasks', ['t1']) + const t1Again = await createWithId(database, tasks, 't1') + expect(t1Again).not.toBe(t1) + expect(observer).toHaveBeenCalledTimes(3) + expect(observer).toHaveBeenLastCalledWith(t1Again) + + await database.write(() => t1Again.destroyPermanently()) + expect(observer).toHaveBeenCalledTimes(4) + expect(observer).toHaveBeenLastCalledWith(null) + + expect(error).toHaveBeenCalledTimes(0) + expect(complete).toHaveBeenCalledTimes(0) + subscription.unsubscribe() + }) + it('fetches records that are not cached, including ids containing colons', async () => { + const { database, tasks: originalTasks, cloneDatabase } = mockDatabase() + const id = '12:Quiz:hand' + await createWithId(database, originalTasks, id) + + // simulate reload - the record is in the database, but not in cache + const reloadedDb = await cloneDatabase() + const tasks = reloadedDb.get('mock_tasks') + const findSpy = jest.spyOn(reloadedDb.adapter.underlyingAdapter, 'find') + + const observer = jest.fn() + tasks.observeRecord(id).subscribe(observer) + expect(observer).toHaveBeenCalledTimes(0) + + await new Promise(process.nextTick) + expect(findSpy).toHaveBeenCalledTimes(1) + expect(findSpy.mock.calls[0].slice(0, 2)).toEqual(['mock_tasks', id]) + expect(observer).toHaveBeenCalledTimes(1) + expect(observer.mock.calls[0][0].id).toBe(id) + + // a missing record with such an id is also fine + const observer2 = jest.fn() + tasks.observeRecord('12:Quiz:missing').subscribe(observer2) + await new Promise(process.nextTick) + expect(observer2).toHaveBeenLastCalledWith(null) + }) + it('does not emit twice if the record is created while fetching', async () => { + const { database, tasks, adapter } = mockDatabase() + + let finishFind = null + adapter.find = jest.fn().mockImplementation((table, id, callback) => { + finishFind = () => callback({ value: null }) + }) + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + expect(adapter.find).toHaveBeenCalledTimes(1) + + const t1 = await createWithId(database, tasks, 't1') + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(t1) + + finishFind() + expect(observer).toHaveBeenCalledTimes(1) + }) + it('notifies every subscription when the record is deleted', async () => { + const { database, tasks } = mockDatabase() + const t1 = await createWithId(database, tasks, 't1') + + const observers = [jest.fn(), jest.fn(), jest.fn()] + observers.forEach((observer) => tasks.observeRecord('t1').subscribe(observer)) + observers.forEach((observer) => expect(observer).toHaveBeenLastCalledWith(t1)) + + await database.write(() => t1.markAsDeleted()) + observers.forEach((observer) => { + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(null) + }) + expect(t1._subscribers.length).toBe(0) + }) + it('emits null for records marked as deleted', async () => { + const { database, tasks } = mockDatabase() + const t1 = await createWithId(database, tasks, 't1') + await database.write(() => t1.markAsDeleted()) + expect(tasks._cache.get('t1')).toBe(undefined) + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + + // find() semantics are unchanged + expect((await tasks.find('t1')).syncStatus).toBe('deleted') + }) + it('does not leak listeners when the consumer unsubscribes synchronously', async () => { + const { database, tasks } = mockDatabase() + + // record created later + const observer = jest.fn() + tasks + .observeRecord('t1') + .pipe(filter(Boolean), take(1)) + .subscribe(observer) + await new Promise(process.nextTick) + const t1 = await createWithId(database, tasks, 't1') + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(t1) + expect(t1._subscribers.length).toBe(0) + expect(tasks._subscribers.length).toBe(0) + + // cached record emitted synchronously on subscription + const observer2 = jest.fn() + tasks.observeRecord('t1').pipe(take(1)).subscribe(observer2) + expect(observer2).toHaveBeenLastCalledWith(t1) + expect(t1._subscribers.length).toBe(0) + expect(tasks._subscribers.length).toBe(0) + }) + it('only listens to collection changes while the record is missing', async () => { + const { database, tasks } = mockDatabase() + + const observer = jest.fn() + const subscription = tasks.observeRecord('t1').subscribe(observer) + await new Promise(process.nextTick) + expect(tasks._subscribers.length).toBe(1) + + const t1 = await createWithId(database, tasks, 't1') + expect(observer).toHaveBeenLastCalledWith(t1) + expect(tasks._subscribers.length).toBe(0) + expect(t1._subscribers.length).toBe(1) + + await database.write(() => t1.markAsDeleted()) + expect(observer).toHaveBeenLastCalledWith(null) + expect(tasks._subscribers.length).toBe(1) + expect(t1._subscribers.length).toBe(0) + + subscription.unsubscribe() + expect(tasks._subscribers.length).toBe(0) + }) + it('does not emit a record created and deleted within batched notifications', async () => { + const { database, tasks } = mockDatabase() + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + await new Promise(process.nextTick) + expect(observer).toHaveBeenLastCalledWith(null) + + await database.experimentalBatchNotifications(async () => { + const t1 = await createWithId(database, tasks, 't1') + await database.write(() => t1.destroyPermanently()) + }) + expect(observer).toHaveBeenCalledTimes(1) + }) + it('switches to a record deleted and re-created within batched notifications', async () => { + const { database, tasks } = mockDatabase() + const t1 = await createWithId(database, tasks, 't1') + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + expect(observer).toHaveBeenLastCalledWith(t1) + + let t1Again + await database.experimentalBatchNotifications(async () => { + await database.write(() => t1.destroyPermanently()) + t1Again = await createWithId(database, tasks, 't1') + }) + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(t1Again) + expect(t1._subscribers.length).toBe(0) + expect(t1Again._subscribers.length).toBe(1) + }) + it('emits null if the record is created and deleted while the initial fetch is pending', async () => { + const { database, tasks, adapter } = mockDatabase() + + let finishFind = null + adapter.find = jest.fn().mockImplementation((table, id, callback) => { + finishFind = () => callback({ value: null }) + }) + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + expect(observer).toHaveBeenCalledTimes(0) + + await database.experimentalBatchNotifications(async () => { + const t1 = await createWithId(database, tasks, 't1') + await database.write(() => t1.destroyPermanently()) + }) + expect(observer.mock.calls).toEqual([[null]]) + + finishFind() + expect(observer.mock.calls).toEqual([[null]]) + + // ...and still recovers + const t1Again = await createWithId(database, tasks, 't1') + expect(observer.mock.calls).toEqual([[null], [t1Again]]) + }) + it('ignores the initial fetch result if the record changed in the meantime', async () => { + const { database, tasks, adapter } = mockDatabase() + + let finishFind = null + adapter.find = jest.fn().mockImplementation((table, id, callback) => { + finishFind = () => callback({ value: { id: 't1', _status: 'synced', _changed: '' } }) + }) + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + + const t1 = await createWithId(database, tasks, 't1') + await database.write(() => t1.markAsDeleted()) + expect(observer.mock.calls).toEqual([[t1], [null]]) + + // outdated result of the initial fetch arrives late + finishFind() + expect(observer.mock.calls).toEqual([[t1], [null]]) + }) + it('does not emit an update of a record deleted within batched notifications', async () => { + const { database, tasks } = mockDatabase() + const t1 = await createWithId(database, tasks, 't1') + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + expect(observer.mock.calls).toEqual([[t1]]) + + await database.experimentalBatchNotifications(async () => { + await database.write(() => t1.update()) + await database.write(() => t1.markAsDeleted()) + }) + expect(observer.mock.calls).toEqual([[t1], [null]]) + }) + it('errors for invalid ids', async () => { + const { tasks } = mockDatabase() + + const observer = jest.fn() + const error = jest.fn() + tasks.observeRecord(null).subscribe(observer, error) + + expect(observer).toHaveBeenCalledTimes(0) + expect(error).toHaveBeenCalledTimes(1) + expect(error.mock.calls[0][0].message).toMatch('Invalid record ID') + }) + it('stops observing after unsubscribing', async () => { + const { database, tasks } = mockDatabase() + + const observer = jest.fn() + const subscription = tasks.observeRecord('t1').subscribe(observer) + await new Promise(process.nextTick) + expect(observer).toHaveBeenCalledTimes(1) + + subscription.unsubscribe() + expect(tasks._subscribers.length).toBe(0) + + const t1 = await createWithId(database, tasks, 't1') + await database.write(() => t1.markAsDeleted()) + expect(observer).toHaveBeenCalledTimes(1) + expect(t1._subscribers.length).toBe(0) + }) +}) diff --git a/src/Model/index.js b/src/Model/index.js index 29596fa00..e21f77ce5 100644 --- a/src/Model/index.js +++ b/src/Model/index.js @@ -416,14 +416,16 @@ export default class Model { _notifyChanged(): void { this._getChanges().next(this) - this._subscribers.forEach(([subscriber]) => { + // Iterate over a copy, so that subscribers can safely unsubscribe while being notified + this._subscribers.slice().forEach(([subscriber]) => { subscriber(false) }) } _notifyDestroyed(): void { this._getChanges().complete() - this._subscribers.forEach(([subscriber]) => { + // Iterate over a copy, so that subscribers can safely unsubscribe while being notified + this._subscribers.slice().forEach(([subscriber]) => { subscriber(true) }) } diff --git a/src/Relation/helpers.d.ts b/src/Relation/helpers.d.ts index aa6bf44a4..37c00b4f6 100644 --- a/src/Relation/helpers.d.ts +++ b/src/Relation/helpers.d.ts @@ -5,4 +5,6 @@ import type { Observable } from '../utils/rx' import type Relation from './index' import type Model from '../Model' -export declare function createObservable(relation: Relation): Observable +export declare function createObservable( + relation: Relation, +): Observable diff --git a/src/Relation/helpers.js b/src/Relation/helpers.js index 49fe9838a..a62beda4f 100644 --- a/src/Relation/helpers.js +++ b/src/Relation/helpers.js @@ -9,13 +9,23 @@ import { } from '../utils/rx' import type Relation from './index' -import type Model from '../Model' +import type Model, { RecordId } from '../Model' -const getImmutableObservable = (relation: Relation): Observable => - relation._model.collections - .get(relation._relationTableName) +// Failsafe relations emit `null` (and keep observing) instead of erroring when the related +// record does not exist, so that they recover when the record is created later +const observeRecord = (relation: Relation, id: ?RecordId): Observable => { + const collection = relation._model.collections.get(relation._relationTableName) + if (relation._isFailsafe) { // $FlowFixMe - .findAndObserve(relation.id) + return id ? collection.observeRecord(id) : of$(null) + } + // $FlowFixMe + return collection.findAndObserve(id) +} + +const getImmutableObservable = (relation: Relation): Observable => + // $FlowFixMe + observeRecord(relation, relation.id) const getObservable = (relation: Relation): Observable => relation._model @@ -24,11 +34,7 @@ const getObservable = (relation: Relation): Observable => .pipe( map$((model) => model._getRaw(relation._columnName)), distinctUntilChanged(), - switchMap((id) => - id - ? relation._model.collections.get(relation._relationTableName).findAndObserve(id) - : of$(null), - ), + switchMap((id) => (id ? observeRecord(relation, id) : of$(null))), ) // eslint-disable-next-line diff --git a/src/Relation/index.d.ts b/src/Relation/index.d.ts index e863d2afd..f371a1fc6 100644 --- a/src/Relation/index.d.ts +++ b/src/Relation/index.d.ts @@ -1,4 +1,4 @@ -import { $NonMaybeType, $Exact, $Call } from '../types' +import { $Exact, $Call } from '../types' import type { Observable } from '../utils/rx' import type Model from '../Model' @@ -11,11 +11,17 @@ type ExtractRecordId = ExtractRecordIdNonOptional & ExtractRecordIdOptional export type Options = $Exact<{ isImmutable: boolean + // If true, observing a related record that does not exist emits `null` (and recovers once the + // record is created) instead of erroring. See `Collection.observeRecord` and `@experimentalFailsafe` + failsafe?: boolean }> // Defines a one-to-one relation between two Models (two tables in db) // Do not create this object directly! Use `relation` or `immutableRelation` decorators instead -export default class Relation { +// +// Use a nullable type (e.g. `Relation`) if the relation ID is optional, or if the +// relation is marked with `@experimentalFailsafe` (`Relation` by default) +export default class Relation { // Used by withObservables to differentiate between object types static _wmelonTag: string @@ -23,17 +29,19 @@ export default class Relation { _columnName: ColumnName - _relationTableName: TableName<$NonMaybeType> + _relationTableName: TableName> _isImmutable: boolean + _isFailsafe: boolean + // TODO: FIX TS // @lazy _cachedObservable: Observable constructor( model: Model, - relationTableName: TableName<$NonMaybeType>, + relationTableName: TableName>, columnName: ColumnName, options: Options, ) diff --git a/src/Relation/index.js b/src/Relation/index.js index 0ad850679..84740f36e 100644 --- a/src/Relation/index.js +++ b/src/Relation/index.js @@ -16,6 +16,9 @@ type ExtractRecordId = ExtractRecordIdNonOptional & ExtractRecordIdOptional export type Options = $Exact<{ isImmutable: boolean, + // If true, observing a related record that does not exist emits `null` (and recovers once the + // record is created) instead of erroring. See `Collection.observeRecord` and `@experimentalFailsafe` + failsafe?: boolean, }> // Defines a one-to-one relation between two Models (two tables in db) @@ -32,6 +35,8 @@ export default class Relation { _isImmutable: boolean + _isFailsafe: boolean + @lazy _cachedObservable: Observable = createObservable(this) .pipe(publishReplayLatestWhileConnected) @@ -47,6 +52,7 @@ export default class Relation { this._relationTableName = relationTableName this._columnName = columnName this._isImmutable = options.isImmutable + this._isFailsafe = !!options.failsafe } get id(): $Call { diff --git a/src/Relation/test.js b/src/Relation/test.js index 79e7e8501..48e3ef9e1 100644 --- a/src/Relation/test.js +++ b/src/Relation/test.js @@ -148,6 +148,124 @@ describe('Relation', () => { expect(await relation).toBe(currentRecord) expect(await relation.then((model) => [model])).toEqual([currentRecord]) }) + it('errors when observing a related record that does not exist', async () => { + const { tasks, db } = mockDatabase() + const primary = await db.write(() => + tasks.create((mock) => { + mock.projectId = 'missing' + }), + ) + const relation = new Relation(primary, 'mock_projects', 'project_id', { isImmutable: false }) + + const observer = jest.fn() + const error = jest.fn() + relation.observe().subscribe(observer, error) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(0) + expect(error).toHaveBeenCalledTimes(1) + }) + it('observes a missing related record as null when failsafe, and recovers', async () => { + const { tasks, projects, db } = mockDatabase() + const primary = await db.write(() => + tasks.create((mock) => { + mock.projectId = 'missing' + }), + ) + const relation = new Relation(primary, 'mock_projects', 'project_id', { + isImmutable: false, + failsafe: true, + }) + expect(relation._isFailsafe).toBe(true) + + const observer = jest.fn() + const error = jest.fn() + const complete = jest.fn() + const subscription = relation.observe().subscribe(observer, error, complete) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + + // related record arrives later (e.g. via sync) + const secondary = await db.write(() => + projects.create((mock) => { + mock._raw.id = 'missing' + mock.name = 'foo' + }), + ) + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(secondary) + + await db.write(() => + secondary.update((mock) => { + mock.name = 'bar' + }), + ) + expect(observer).toHaveBeenCalledTimes(3) + expect(observer).toHaveBeenLastCalledWith(secondary) + + // related record is deleted + await db.write(() => secondary.markAsDeleted()) + expect(observer).toHaveBeenCalledTimes(4) + expect(observer).toHaveBeenLastCalledWith(null) + + // relation is re-pointed at another record + const secondary2 = await db.write(() => projects.create()) + await db.write(() => + primary.update((mock) => { + mock.projectId = secondary2.id + }), + ) + expect(observer).toHaveBeenLastCalledWith(secondary2) + + expect(error).toHaveBeenCalledTimes(0) + expect(complete).toHaveBeenCalledTimes(0) + subscription.unsubscribe() + }) + it('observes a missing immutable related record as null when failsafe, and recovers', async () => { + const { tasks, comments, db } = mockDatabase() + const primary = await db.write(() => + comments.create((mock) => { + mock.task.id = 'missing' + }), + ) + const relation = new Relation(primary, 'mock_tasks', 'task_id', { + isImmutable: true, + failsafe: true, + }) + + const observer = jest.fn() + const subscription = relation.observe().subscribe(observer) + await new Promise(process.nextTick) + expect(observer).toHaveBeenLastCalledWith(null) + + const secondary = await db.write(() => + tasks.create((mock) => { + mock._raw.id = 'missing' + }), + ) + expect(observer).toHaveBeenLastCalledWith(secondary) + subscription.unsubscribe() + }) + it('observes a failsafe immutable relation with a null id as null', async () => { + const { tasks, db } = mockDatabase() + const primary = await db.write(() => tasks.create()) + const relation = new Relation(primary, 'mock_project_sections', 'project_section_id', { + isImmutable: true, + failsafe: true, + }) + expect(relation.id).toBe(null) + + const observer = jest.fn() + const error = jest.fn() + relation.observe().subscribe(observer, error) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + expect(error).toHaveBeenCalledTimes(0) + }) it('caches observable', () => { const { tasks } = mockDatabase() const model = new MockTask(tasks, {}) diff --git a/src/decorators/experimentalFailsafe/index.js b/src/decorators/experimentalFailsafe/index.js index 7fc40df69..6ae6bf65d 100644 --- a/src/decorators/experimentalFailsafe/index.js +++ b/src/decorators/experimentalFailsafe/index.js @@ -1,7 +1,7 @@ // @flow import { isObj } from '../../utils/fp' -import { catchError, of } from '../../utils/rx' +import { catchError, map, of } from '../../utils/rx' type FailsafeDecorator = ( fallback?: any, @@ -29,6 +29,14 @@ const failsafe: FailsafeDecorator = const originalFetch = value.fetch const originalObserve = value.observe + // A Relation can recover from a missing related record: when marked as failsafe, its + // `observe()` emits `null` while the record does not exist (and the record once it + // is created) instead of erroring — which, if caught below, would end the observation + const isRelation = !!value.constructor && value.constructor._wmelonTag === 'relation' + if (isRelation) { + value._isFailsafe = true + } + if (typeof originalFetch === 'function') { value.fetch = function fetch(...args): any { const result = originalFetch.apply(value, args) @@ -43,7 +51,15 @@ const failsafe: FailsafeDecorator = value.observe = function observe(...args): any { const result = originalObserve.apply(value, args) if (isObj(result) && typeof result.pipe === 'function') { - return result.pipe(catchError(() => of(fallback))) + const operators = [catchError(() => of(fallback))] + if (isRelation) { + // Missing related record is emitted as `null` — map it to the fallback value + // (a relation whose ID is null legitimately emits `null`, so leave that alone) + operators.unshift( + map((record) => (record === null && value.id ? fallback : record)), + ) + } + return result.pipe(...operators) } return result } diff --git a/src/decorators/experimentalFailsafe/test.js b/src/decorators/experimentalFailsafe/test.js new file mode 100644 index 000000000..2d4acb1c5 --- /dev/null +++ b/src/decorators/experimentalFailsafe/test.js @@ -0,0 +1,157 @@ +import { of, throwError } from 'rxjs' +import { MockTask, mockDatabase } from '../../__tests__/testModels' + +import experimentalFailsafe from './index' +import relation from '../relation' +import immutableRelation from '../immutableRelation' + +const makeProxy = (task, fallback) => { + class TaskProxy { + asModel = task + + @experimentalFailsafe(fallback) + @relation('mock_projects', 'project_id') + project + + @experimentalFailsafe(fallback) + @immutableRelation('mock_project_sections', 'project_section_id') + projectSection + } + return new TaskProxy() +} + +describe('decorators/experimentalFailsafe', () => { + it('marks relations as failsafe', () => { + const { tasks } = mockDatabase() + const proxy = makeProxy(new MockTask(tasks, { project_id: 'p1' })) + expect(proxy.project._isFailsafe).toBe(true) + expect(proxy.projectSection._isFailsafe).toBe(true) + expect(new MockTask(tasks, {}).project._isFailsafe).toBe(false) + }) + it('fetches missing related records as the fallback value', async () => { + const { tasks, projects, db } = mockDatabase() + const task = await db.write(() => + tasks.create((mock) => { + mock.projectId = 'missing' + }), + ) + expect(await makeProxy(task).project.fetch()).toBe(undefined) + expect(await makeProxy(task, 'fallback').project.fetch()).toBe('fallback') + expect(await makeProxy(task).projectSection.fetch()).toBe(null) // null id + + const project = await db.write(() => + projects.create((mock) => { + mock._raw.id = 'missing' + }), + ) + expect(await makeProxy(task).project.fetch()).toBe(project) + }) + it('observes missing related records as the fallback value, and recovers', async () => { + const { tasks, projects, db } = mockDatabase() + const task = await db.write(() => + tasks.create((mock) => { + mock.projectId = 'missing' + }), + ) + const proxy = makeProxy(task, 'fallback') + + const observer = jest.fn() + const error = jest.fn() + const complete = jest.fn() + const subscription = proxy.project.observe().subscribe(observer, error, complete) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith('fallback') + + // related record arrives later + const project = await db.write(() => + projects.create((mock) => { + mock._raw.id = 'missing' + }), + ) + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(project) + + // ...and is deleted again + await db.write(() => project.markAsDeleted()) + expect(observer).toHaveBeenCalledTimes(3) + expect(observer).toHaveBeenLastCalledWith('fallback') + + // a null relation id still yields null, not the fallback + await db.write(() => + task.update((mock) => { + mock.projectId = null + }), + ) + expect(observer).toHaveBeenLastCalledWith(null) + + expect(error).toHaveBeenCalledTimes(0) + expect(complete).toHaveBeenCalledTimes(0) + subscription.unsubscribe() + }) + it('observes missing immutable related records as the fallback value, and recovers', async () => { + const { tasks, projectSections, db } = mockDatabase() + const task = await db.write(() => + tasks.create((mock) => { + mock._setRaw('project_section_id', 'missing') + }), + ) + const proxy = makeProxy(task) + + const observer = jest.fn() + const subscription = proxy.projectSection.observe().subscribe(observer) + await new Promise(process.nextTick) + expect(observer).toHaveBeenLastCalledWith(undefined) + + const section = await db.write(() => + projectSections.create((mock) => { + mock._raw.id = 'missing' + }), + ) + expect(observer).toHaveBeenLastCalledWith(section) + subscription.unsubscribe() + }) + it('observes an immutable relation with a null id as null, not the fallback', async () => { + const { tasks, db } = mockDatabase() + const task = await db.write(() => tasks.create()) + const proxy = makeProxy(task, 'fallback') + + const observer = jest.fn() + const complete = jest.fn() + proxy.projectSection.observe().subscribe(observer, undefined, complete) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + }) + it('still catches errors of arbitrary observable-like values', async () => { + class Proxy { + @experimentalFailsafe('fallback') + get broken() { + return { + fetch: () => Promise.reject(new Error('nope')), + observe: () => throwError(() => new Error('nope')), + } + } + + @experimentalFailsafe('fallback') + get working() { + return { + fetch: () => Promise.resolve('value'), + observe: () => of('value'), + } + } + } + const proxy = new Proxy() + + expect(await proxy.broken.fetch()).toBe('fallback') + expect(await proxy.working.fetch()).toBe('value') + + const observer = jest.fn() + proxy.broken.observe().subscribe(observer) + expect(observer).toHaveBeenLastCalledWith('fallback') + proxy.working.observe().subscribe(observer) + expect(observer).toHaveBeenLastCalledWith('value') + }) +}) diff --git a/src/react/withObservables/withObservables.test.js b/src/react/withObservables/withObservables.test.js index 6c6fe5469..9a6533cc1 100644 --- a/src/react/withObservables/withObservables.test.js +++ b/src/react/withObservables/withObservables.test.js @@ -1,4 +1,7 @@ import * as React from 'react' +import * as TestRenderer from 'react-test-renderer' +import { Subject } from 'rxjs' +import { mockDatabase } from '../../__tests__/testModels' import withObservables from './index' describe('withObservables', () => { @@ -13,4 +16,52 @@ describe('withObservables', () => { const WrappedComponent = withObservables([], getObservables)(A) expect(WrappedComponent.nonReactProp).toBe(A.nonReactProp) }) + it('renders a null prop without throwing, and re-renders when the record arrives', async () => { + const { database, tasks } = mockDatabase() + + // eslint-disable-next-line react/prop-types + const Task = ({ task }) => {task ? `task:${task.id}` : 'missing'} + const EnhancedTask = withObservables(['taskId'], ({ taskId }) => ({ + task: tasks.observeRecord(taskId), + }))(Task) + + let renderer + await TestRenderer.act(async () => { + renderer = TestRenderer.create() + }) + expect(renderer.root.findByType('span').children).toEqual(['missing']) + + await TestRenderer.act(async () => { + await database.write(() => + tasks.create((task) => { + task._raw.id = 't1' + }), + ) + }) + expect(renderer.root.findByType('span').children).toEqual(['task:t1']) + + await TestRenderer.act(async () => { + await database.write(async () => { + const task = await tasks.find('t1') + await task.markAsDeleted() + }) + }) + expect(renderer.root.findByType('span').children).toEqual(['missing']) + + renderer.unmount() + }) + it('rethrows observable errors in render', () => { + const subject = new Subject() + const Base = () => ok + const Enhanced = withObservables([], () => ({ value: subject }))(Base) + const consoleError = jest.spyOn(console, 'error').mockImplementation(() => {}) + + expect(() => { + TestRenderer.act(() => { + TestRenderer.create() + subject.error(new Error('boom')) + }) + }).toThrow('boom') + consoleError.mockRestore() + }) })