Add Collection.observeRecord for records that may be missing

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

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`.

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`.
 

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<User | undefined>`), 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)

src/Collection/index.d.ts

@@ -41,6 +41,11 @@ export default class Collection<Record extends Model> {
   // (with the same semantics as when calling `model.observe()`)
   findAndObserve(id: RecordId): Observable<Record>
 
+  // 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<Record | null>
+
   // Query records of this type
   query: ArrayOrSpreadFn<Clause, Query<Record>>
 
@@ -87,6 +92,9 @@ export default class Collection<Record extends Model> {
   // Fetches exactly one record (See: Collection.find)
   _fetchRecord(id: RecordId, callback: ResultCallback<Record>): void
 
+  // Fetches exactly one record, or `null` if it does not exist (See: Collection.observeRecord)
+  _fetchRecordIfExists(id: RecordId, callback: ResultCallback<Record | null>): void
+
   _applyChangesToCache(operations: CollectionChangeSet<Record>): void
 
   _notify(operations: CollectionChangeSet<Record>): void

src/Collection/index.js

@@ -116,6 +116,130 @@ export default class Collection<Record: Model> {
     })
   }
 
+  /**
+   * 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<?Record> {
+    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<Clause, Query<Record>>  */
   /**
    * Returns a `Query` with conditions given.
@@ -217,6 +341,18 @@ export default class Collection<Record: Model> {
 
   // Fetches exactly one record (See: Collection.find)
   _fetchRecord(id: RecordId, callback: ResultCallback<Record>): 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<?Record>): void {
     if (typeof id !== 'string') {
       callback({ error: new Error(`Invalid record ID ${this.table}#${id}`) })
       return
@@ -231,10 +367,10 @@ export default class Collection<Record: Model> {
 
     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<Record: Model> {
     ]): 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<Record>): void => {