feat(firestore): add support for local source snapshot listeners

packages/firestore/__tests__/firestore.test.ts

@@ -7,6 +7,7 @@ import FirebaseModule from '../../app/lib/internal/FirebaseModule';
 import Query from '../lib/FirestoreQuery';
 // @ts-ignore test
 import FirestoreDocumentSnapshot from '../lib/FirestoreDocumentSnapshot';
+import { parseSnapshotArgs } from '../lib/utils';
 // @ts-ignore test
 import * as nativeModule from '@react-native-firebase/app/dist/module/internal/nativeModuleAndroidIos';
 
@@ -498,6 +499,35 @@ describe('Firestore', function () {
         });
       });
     });
+
+    describe('onSnapshot()', function () {
+      it("accepts { source: 'cache' } listener options", function () {
+        const parsed = parseSnapshotArgs([{ source: 'cache' }, () => {}]);
+
+        expect(parsed.snapshotListenOptions).toEqual({
+          includeMetadataChanges: false,
+          source: 'cache',
+        });
+      });
+
+      it("accepts { source: 'default', includeMetadataChanges: true } listener options", function () {
+        const parsed = parseSnapshotArgs([
+          { source: 'default', includeMetadataChanges: true },
+          () => {},
+        ]);
+
+        expect(parsed.snapshotListenOptions).toEqual({
+          includeMetadataChanges: true,
+          source: 'default',
+        });
+      });
+
+      it("throws for unsupported listener source value 'server'", function () {
+        expect(() =>
+          parseSnapshotArgs([{ source: 'server' as 'default' | 'cache' }, () => {}]),
+        ).toThrow("'options' SnapshotOptions.source must be one of 'default' or 'cache'.");
+      });
+    });
   });
 
   describe('modular', function () {

packages/firestore/android/src/reactnative/java/io/invertase/firebase/firestore/ReactNativeFirebaseFirestoreCollectionModule.java

@@ -334,6 +334,7 @@ private void handleQueryOnSnapshot(
       int listenerId,
       ReadableMap listenerOptions) {
     MetadataChanges metadataChanges;
+    SnapshotListenOptions.Builder snapshotListenOptionsBuilder = new SnapshotListenOptions.Builder();
 
     if (listenerOptions != null
         && listenerOptions.hasKey("includeMetadataChanges")
@@ -342,6 +343,15 @@ private void handleQueryOnSnapshot(
     } else {
       metadataChanges = MetadataChanges.EXCLUDE;
     }
+    snapshotListenOptionsBuilder.setMetadataChanges(metadataChanges);
+
+    if (listenerOptions != null
+        && listenerOptions.hasKey("source")
+        && "cache".equals(listenerOptions.getString("source"))) {
+      snapshotListenOptionsBuilder.setSource(ListenSource.CACHE);
+    } else {
+      snapshotListenOptionsBuilder.setSource(ListenSource.DEFAULT);
+    }
 
     final EventListener<QuerySnapshot> listener =
         (querySnapshot, exception) -> {
@@ -358,7 +368,7 @@ private void handleQueryOnSnapshot(
         };
 
     ListenerRegistration listenerRegistration =
-        firestoreQuery.query.addSnapshotListener(metadataChanges, listener);
+        firestoreQuery.query.addSnapshotListener(snapshotListenOptionsBuilder.build(), listener);
 
     collectionSnapshotListeners.put(listenerId, listenerRegistration);
   }

packages/firestore/android/src/reactnative/java/io/invertase/firebase/firestore/ReactNativeFirebaseFirestoreDocumentModule.java

@@ -78,18 +78,26 @@ public void documentOnSnapshot(
           }
         };
 
-    MetadataChanges metadataChanges;
+    SnapshotListenOptions.Builder snapshotListenOptionsBuilder = new SnapshotListenOptions.Builder();
 
     if (listenerOptions != null
         && listenerOptions.hasKey("includeMetadataChanges")
         && listenerOptions.getBoolean("includeMetadataChanges")) {
-      metadataChanges = MetadataChanges.INCLUDE;
+      snapshotListenOptionsBuilder.setMetadataChanges(MetadataChanges.INCLUDE);
     } else {
-      metadataChanges = MetadataChanges.EXCLUDE;
+      snapshotListenOptionsBuilder.setMetadataChanges(MetadataChanges.EXCLUDE);
+    }
+
+    if (listenerOptions != null
+        && listenerOptions.hasKey("source")
+        && "cache".equals(listenerOptions.getString("source"))) {
+      snapshotListenOptionsBuilder.setSource(ListenSource.CACHE);
+    } else {
+      snapshotListenOptionsBuilder.setSource(ListenSource.DEFAULT);
     }
 
     ListenerRegistration listenerRegistration =
-        documentReference.addSnapshotListener(metadataChanges, listener);
+        documentReference.addSnapshotListener(snapshotListenOptionsBuilder.build(), listener);
 
     documentSnapshotListeners.put(listenerId, listenerRegistration);
   }

packages/firestore/e2e/DocumentReference/onSnapshot.e2e.js

@@ -16,7 +16,7 @@
  */
 const COLLECTION = 'firestore';
 const NO_RULE_COLLECTION = 'no_rules';
-const { wipe } = require('../helpers');
+const { wipe, setDocumentOutOfBand } = require('../helpers');
 
 describe('firestore().doc().onSnapshot()', function () {
   before(function () {
@@ -305,6 +305,108 @@ describe('firestore().doc().onSnapshot()', function () {
       }
     });
 
+    it("throws if SnapshotListenerOptions.source is invalid ('server')", function () {
+      try {
+        firebase.firestore().doc(`${NO_RULE_COLLECTION}/nope`).onSnapshot({
+          source: 'server',
+        });
+        return Promise.reject(new Error('Did not throw an Error.'));
+      } catch (error) {
+        error.message.should.containEql(
+          "'options' SnapshotOptions.source must be one of 'default' or 'cache'",
+        );
+        return Promise.resolve();
+      }
+    });
+
+    it('accepts source-only SnapshotListenerOptions', async function () {
+      if (Platform.other) {
+        return;
+      }
+      const callback = sinon.spy();
+      const unsub = firebase.firestore().doc(`${COLLECTION}/source-only`).onSnapshot(
+        {
+          source: 'cache',
+        },
+        callback,
+      );
+
+      await Utils.spyToBeCalledOnceAsync(callback);
+      unsub();
+    });
+
+    it('accepts source + includeMetadataChanges SnapshotListenerOptions', async function () {
+      if (Platform.other) {
+        return;
+      }
+      const callback = sinon.spy();
+      const unsub = firebase.firestore().doc(`${COLLECTION}/source-with-metadata`).onSnapshot(
+        {
+          source: 'default',
+          includeMetadataChanges: true,
+        },
+        callback,
+      );
+
+      await Utils.spyToBeCalledOnceAsync(callback);
+      unsub();
+    });
+
+    it('cache source listeners ignore out-of-band server writes', async function () {
+      if (Platform.other) {
+        return;
+      }
+
+      const docPath = `${COLLECTION}/${Utils.randString(12, '#aA')}`;
+      const docRef = firebase.firestore().doc(docPath);
+      await docRef.set({ value: 1 });
+      await docRef.get();
+
+      const callback = sinon.spy();
+      const unsub = docRef.onSnapshot({ source: 'cache' }, callback);
+      try {
+        await Utils.spyToBeCalledOnceAsync(callback);
+
+        await setDocumentOutOfBand(docPath, { value: 2 });
+        await Utils.sleep(1500);
+        callback.should.be.callCount(1);
+
+        await docRef.set({ value: 3 });
+        await Utils.spyToBeCalledTimesAsync(callback, 2);
+        callback.args[1][0].get('value').should.equal(3);
+      } finally {
+        unsub();
+      }
+    });
+
+    it('default source listeners receive out-of-band server writes', async function () {
+      if (Platform.other) {
+        return;
+      }
+
+      const docPath = `${COLLECTION}/${Utils.randString(12, '#aA')}`;
+      const docRef = firebase.firestore().doc(docPath);
+      await docRef.set({ value: 1 });
+      await docRef.get();
+
+      const callback = sinon.spy();
+      const unsub = docRef.onSnapshot(
+        { source: 'default', includeMetadataChanges: true },
+        callback,
+      );
+      try {
+        await Utils.spyToBeCalledOnceAsync(callback);
+
+        await setDocumentOutOfBand(docPath, { value: 2 });
+        await Utils.spyToBeCalledTimesAsync(callback, 2, 8000);
+
+        const latestSnapshot = callback.args[callback.callCount - 1][0];
+        latestSnapshot.get('value').should.equal(2);
+      } finally {
+        unsub();
+      }
+    });
+
     it('throws if next callback is invalid', function () {
       try {
         firebase.firestore().doc(`${NO_RULE_COLLECTION}/nope`).onSnapshot({
@@ -616,6 +718,58 @@ describe('firestore().doc().onSnapshot()', function () {
       }
     });
 
+    it("throws if SnapshotListenerOptions.source is invalid ('server')", function () {
+      const { getFirestore, doc, onSnapshot } = firestoreModular;
+      try {
+        onSnapshot(doc(getFirestore(), `${NO_RULE_COLLECTION}/nope`), {
+          source: 'server',
+        });
+        return Promise.reject(new Error('Did not throw an Error.'));
+      } catch (error) {
+        error.message.should.containEql(
+          "'options' SnapshotOptions.source must be one of 'default' or 'cache'",
+        );
+        return Promise.resolve();
+      }
+    });
+
+    it('accepts source-only SnapshotListenerOptions', async function () {
+      if (Platform.other) {
+        return;
+      }
+      const { getFirestore, doc, onSnapshot } = firestoreModular;
+      const callback = sinon.spy();
+      const unsub = onSnapshot(
+        doc(getFirestore(), `${COLLECTION}/mod-source-only`),
+        {
+          source: 'cache',
+        },
+        callback,
+      );
+
+      await Utils.spyToBeCalledOnceAsync(callback);
+      unsub();
+    });
+
+    it('accepts source + includeMetadataChanges SnapshotListenerOptions', async function () {
+      if (Platform.other) {
+        return;
+      }
+      const { getFirestore, doc, onSnapshot } = firestoreModular;
+      const callback = sinon.spy();
+      const unsub = onSnapshot(
+        doc(getFirestore(), `${COLLECTION}/mod-source-with-metadata`),
+        {
+          source: 'default',
+          includeMetadataChanges: true,
+        },
+        callback,
+      );
+
+      await Utils.spyToBeCalledOnceAsync(callback);
+      unsub();
+    });
+
     it('throws if next callback is invalid', function () {
       const { getFirestore, doc, onSnapshot } = firestoreModular;
       try {

packages/firestore/e2e/Query/onSnapshot.e2e.js

@@ -14,7 +14,7 @@
  * limitations under the License.
  *
  */
-const { wipe } = require('../helpers');
+const { wipe, setDocumentOutOfBand } = require('../helpers');
 const COLLECTION = 'firestore';
 const NO_RULE_COLLECTION = 'no_rules';
 
@@ -319,6 +319,69 @@ describe('firestore().collection().onSnapshot()', function () {
       }
     });
 
+    it("throws if SnapshotListenerOptions.source is invalid ('server')", function () {
+      try {
+        firebase.firestore().collection(NO_RULE_COLLECTION).onSnapshot({
+          source: 'server',
+        });
+        return Promise.reject(new Error('Did not throw an Error.'));
+      } catch (error) {
+        error.message.should.containEql(
+          "'options' SnapshotOptions.source must be one of 'default' or 'cache'",
+        );
+        return Promise.resolve();
+      }
+    });
+
+    it('cache source query listeners ignore out-of-band server writes', async function () {
+      if (Platform.other) {
+        return;
+      }
+
+      const collectionPath = `${COLLECTION}/${Utils.randString(12, '#aA')}/cache-source`;
+      const colRef = firebase.firestore().collection(collectionPath);
+      await colRef.doc('one').set({ enabled: true });
+      await colRef.get();
+
+      const callback = sinon.spy();
+      const unsub = colRef.onSnapshot({ source: 'cache' }, callback);
+      try {
+        await Utils.spyToBeCalledOnceAsync(callback);
+        await setDocumentOutOfBand(`${collectionPath}/server-write`, { enabled: false });
+        await Utils.sleep(1500);
+        callback.should.be.callCount(1);
+
+        await colRef.doc('local-write').set({ enabled: true });
+        await Utils.spyToBeCalledTimesAsync(callback, 2);
+      } finally {
+        unsub();
+      }
+    });
+
+    it('default source query listeners receive out-of-band server writes', async function () {
+      if (Platform.other) {
+        return;
+      }
+
+      const collectionPath = `${COLLECTION}/${Utils.randString(12, '#aA')}/cache-source-meta`;
+      const colRef = firebase.firestore().collection(collectionPath);
+      await colRef.doc('one').set({ enabled: true });
+      await colRef.get();
+
+      const callback = sinon.spy();
+      const unsub = colRef.onSnapshot(
+        { source: 'default', includeMetadataChanges: true },
+        callback,
+      );
+      try {
+        await Utils.spyToBeCalledOnceAsync(callback);
+        await setDocumentOutOfBand(`${collectionPath}/server-write`, { enabled: false });
+        await Utils.spyToBeCalledTimesAsync(callback, 2, 8000);
+      } finally {
+        unsub();
+      }
+    });
+
     it('throws if next callback is invalid', function () {
       try {
         firebase.firestore().collection(NO_RULE_COLLECTION).onSnapshot({
@@ -637,6 +700,21 @@ describe('firestore().collection().onSnapshot()', function () {
       }
     });
 
+    it("throws if SnapshotListenerOptions.source is invalid ('server')", function () {
+      const { getFirestore, collection, onSnapshot } = firestoreModular;
+      try {
+        onSnapshot(collection(getFirestore(), NO_RULE_COLLECTION), {
+          source: 'server',
+        });
+        return Promise.reject(new Error('Did not throw an Error.'));
+      } catch (error) {
+        error.message.should.containEql(
+          "'options' SnapshotOptions.source must be one of 'default' or 'cache'",
+        );
+        return Promise.resolve();
+      }
+    });
+
     it('throws if next callback is invalid', function () {
       const { getFirestore, collection, onSnapshot } = firestoreModular;
       try {