First working impl

Firestore/Example/Tests/Integration/API/FIRDatabaseTests.mm

@@ -1402,4 +1402,56 @@ - (void)testCanRemoveListenerAfterShutdown {
   [listenerRegistration remove];
 }
 
+- (void)testWaitForPendingWritesCompletes {
+  FIRDocumentReference *doc = [self documentRef];
+  FIRFirestore *firestore = doc.firestore;
+
+  [self disableNetwork];
+
+  [doc setData:@{@"foo" : @"bar"}];
+  [firestore waitForPendingWritesWithCompletion:
+                 [self completionForExpectationWithName:@"Wait for pending writes"]];
+
+  [firestore enableNetworkWithCompletion:[self completionForExpectationWithName:@"Enable network"]];
+  [self awaitExpectations];
+}
+
+- (void)testWaitForPendingWritesFailsWhenUserChanges {
+  FIRApp *app = testutil::AppForUnitTesting(util::MakeString([FSTIntegrationTestCase projectID]));
+  FIRFirestore *firestore = [self firestoreWithApp:app];
+
+  [firestore
+      disableNetworkWithCompletion:[self completionForExpectationWithName:@"Disable network"]];
+  [self awaitExpectations];
+
+  // Writes to local to prevent immediate call to completion of waitForPendingWrites.
+  NSDictionary<NSString *, id> *data =
+      @{@"owner" : @{@"name" : @"Andy", @"email" : @"abc@xyz.com"}};
+  [[firestore documentWithPath:@"abc/123"] setData:data];
+
+  XCTestExpectation *expectation = [self expectationWithDescription:@"waitForPendingWrites"];
+  [firestore waitForPendingWritesWithCompletion:^(NSError *_Nullable error) {
+    XCTAssertNotNil(error);
+    XCTAssertEqualObjects(error.domain, FIRFirestoreErrorDomain);
+    XCTAssertEqual(error.code, FIRFirestoreErrorCodeCancelled);
+    [expectation fulfill];
+  }];
+
+  [self triggerUserChangeWithUid:@"user-to-fail-pending-writes"];
+  [self awaitExpectations];
+}
+
+- (void)testWaitForPendingWritesCompletesWhenOfflineIfNoPending {
+  FIRDocumentReference *doc = [self documentRef];
+  FIRFirestore *firestore = doc.firestore;
+
+  [firestore
+      disableNetworkWithCompletion:[self completionForExpectationWithName:@"Disable network"]];
+  [self awaitExpectations];
+
+  [firestore waitForPendingWritesWithCompletion:
+                 [self completionForExpectationWithName:@"Wait for pending writes"]];
+  [self awaitExpectations];
+}
+
 @end

Firestore/Example/Tests/Local/FSTLevelDBMutationQueueTests.mm

@@ -87,14 +87,14 @@ - (void)setUp {
 
 - (void)testLoadNextBatchID_zeroWhenTotallyEmpty {
   // Initial seek is invalid
-  XCTAssertEqual(LoadNextBatchIdFromDb(_db.ptr), 0);
+  XCTAssertEqual(LoadNextBatchIdFromDb(_db.ptr), 1);
 }
 
 - (void)testLoadNextBatchID_zeroWhenNoMutations {
   // Initial seek finds no mutations
   [self setDummyValueForKey:MutationLikeKey("mutationr", "foo", 20)];
   [self setDummyValueForKey:MutationLikeKey("mutationsa", "foo", 10)];
-  XCTAssertEqual(LoadNextBatchIdFromDb(_db.ptr), 0);
+  XCTAssertEqual(LoadNextBatchIdFromDb(_db.ptr), 1);
 }
 
 - (void)testLoadNextBatchID_findsSingleRow {

Firestore/Example/Tests/Local/FSTLocalStoreTests.mm

@@ -1171,6 +1171,24 @@ - (void)testHandlesPatchMutationWithTransformThenRemoteEvent {
   FSTAssertChanged(@[ FSTTestDoc("foo/bar", 1, @{@"sum" : @1}, DocumentState::kLocalMutations) ]);
 }
 
+- (void)testGetHighestUnacknowledgeBatchId {
+  if ([self isTestBaseClass]) return;
+
+  XCTAssertEqual(-1, [self.localStore getHighestUnacknowledgedBatchId]);
+
+  [self writeMutation:FSTTestSetMutation(@"foo/bar", @{@"abc" : @123})];
+  XCTAssertEqual(1, [self.localStore getHighestUnacknowledgedBatchId]);
+
+  [self writeMutation:FSTTestPatchMutation("foo/bar", @{@"abc" : @321}, {})];
+  XCTAssertEqual(2, [self.localStore getHighestUnacknowledgedBatchId]);
+
+  [self acknowledgeMutationWithVersion:1];
+  XCTAssertEqual(2, [self.localStore getHighestUnacknowledgedBatchId]);
+
+  [self rejectMutation];
+  XCTAssertEqual(-1, [self.localStore getHighestUnacknowledgedBatchId]);
+}
+
 @end
 
 NS_ASSUME_NONNULL_END

Firestore/Example/Tests/Util/FSTIntegrationTestCase.h

@@ -55,7 +55,11 @@ extern "C" {
 /** Returns a new Firestore connected to the project with the given projectID. */
 - (FIRFirestore *)firestoreWithProjectID:(NSString *)projectID;
 
-/** Returns a new Firestore connected to the project with the given app. */
+- (void)triggerUserChangeWithUid:(NSString *)uid;
+
+/**
+ * Returns a new Firestore connected to the project with the given app.
+ */
 - (FIRFirestore *)firestoreWithApp:(FIRApp *)app;
 
 /** Synchronously shuts down the given firestore. */

Firestore/Example/Tests/Util/FSTIntegrationTestCase.mm

@@ -33,7 +33,9 @@
 #include <string>
 #include <utility>
 
+#include "Firestore/core/src/firebase/firestore/auth/credentials_provider.h"
 #include "Firestore/core/src/firebase/firestore/auth/empty_credentials_provider.h"
+#include "Firestore/core/src/firebase/firestore/auth/user.h"
 #include "Firestore/core/src/firebase/firestore/model/database_id.h"
 #include "Firestore/core/src/firebase/firestore/remote/grpc_connection.h"
 #include "Firestore/core/src/firebase/firestore/util/autoid.h"
@@ -55,8 +57,10 @@
 #include "Firestore/core/src/firebase/firestore/util/executor_libdispatch.h"
 
 namespace util = firebase::firestore::util;
+using firebase::firestore::auth::CredentialChangeListener;
 using firebase::firestore::auth::CredentialsProvider;
 using firebase::firestore::auth::EmptyCredentialsProvider;
+using firebase::firestore::auth::User;
 using firebase::firestore::model::DatabaseId;
 using firebase::firestore::testutil::AppForUnitTesting;
 using firebase::firestore::remote::GrpcConnection;
@@ -79,17 +83,42 @@
 
 static bool runningAgainstEmulator = false;
 
+// Behaves the same as `EmptyCredentialsProvider` except it can also trigger a user
+// change.
+class MockCredentialsProvider : public EmptyCredentialsProvider {
+ public:
+  void SetCredentialChangeListener(CredentialChangeListener changeListener) override {
+    if (changeListener) {
+      listener_ = std::move(changeListener);
+      listener_(User::Unauthenticated());
+    }
+  }
+
+  void ChangeUser(NSString *new_id) {
+    if (listener_) {
+      listener_(firebase::firestore::auth::User::FromUid(new_id));
+    }
+  }
+
+ private:
+  CredentialChangeListener listener_;
+};
+
 @implementation FSTIntegrationTestCase {
   NSMutableArray<FIRFirestore *> *_firestores;
+  std::shared_ptr<MockCredentialsProvider> _mockCredProdiver;
 }
 
 - (void)setUp {
   [super setUp];
 
+  _mockCredProdiver = std::make_shared<MockCredentialsProvider>();
+
   [self clearPersistence];
   [self primeBackend];
 
   _firestores = [NSMutableArray array];
+
   self.db = [self firestore];
   self.eventAccumulator = [FSTEventAccumulator accumulatorForTest:self];
 }
@@ -226,22 +255,23 @@ - (FIRFirestore *)firestoreWithApp:(FIRApp *)app {
 
   FIRSetLoggerLevel(FIRLoggerLevelDebug);
 
-  std::unique_ptr<CredentialsProvider> credentials_provider =
-      absl::make_unique<firebase::firestore::auth::EmptyCredentialsProvider>();
   std::string projectID = util::MakeString(app.options.projectID);
-  FIRFirestore *firestore =
-      [[FIRFirestore alloc] initWithDatabaseID:DatabaseId(projectID)
-                                persistenceKey:util::MakeString(persistenceKey)
-                           credentialsProvider:std::move(credentials_provider)
-                                   workerQueue:std::move(workerQueue)
-                                   firebaseApp:app
-                              instanceRegistry:nil];
-
+  FIRFirestore *firestore = nil;
+  firestore = [[FIRFirestore alloc] initWithDatabaseID:DatabaseId(projectID)
+                                        persistenceKey:util::MakeString(persistenceKey)
+                                   credentialsProvider:_mockCredProdiver
+                                           workerQueue:std::move(workerQueue)
+                                           firebaseApp:app
+                                      instanceRegistry:nil];
   firestore.settings = [FSTIntegrationTestCase settings];
   [_firestores addObject:firestore];
   return firestore;
 }
 
+- (void)triggerUserChangeWithUid:(NSString *)uid {
+  _mockCredProdiver->ChangeUser(uid);
+}
+
 - (void)primeBackend {
   static dispatch_once_t onceToken;
   dispatch_once(&onceToken, ^{

Firestore/Source/API/FIRFirestore+Internal.h

@@ -88,6 +88,20 @@ NS_ASSUME_NONNULL_BEGIN
 - (void)shutdownWithCompletion:(nullable void (^)(NSError *_Nullable error))completion
     NS_SWIFT_NAME(shutdown(completion:));
 
+/**
+ * Waits until all currently pending writes for the active user have been acknowledged by the
+ * backend.
+ *
+ * The passed in completion is called immediately if there are no outstanding writes.
+ * Otherwise, the callback waits for all previously issued writes (including those written in a
+ * previous app session), but it does not wait for writes that were added after the method is
+ * called. If you wish to wait for additional writes, you have to call `waitForPendingWrites` again.
+ *
+ * Any outstanding `waitForPendingWrites` callbacks are called with error during user changes.
+ */
+- (void)waitForPendingWrites:(nullable void (^)(NSError *_Nullable error))completion
+    NS_SWIFT_NAME(waitForPendingWrites(completion:));
+
 - (void)shutdownInternalWithCompletion:(nullable void (^)(NSError *_Nullable error))completion;
 
 - (const std::shared_ptr<util::AsyncQueue> &)workerQueue;

Firestore/Source/API/FIRFirestore.mm

@@ -194,6 +194,10 @@ - (FIRWriteBatch *)batch {
                                          writeBatch:_firestore->GetBatch()];
 }
 
+- (void)waitForPendingWritesWithCompletion:(void (^)(NSError *_Nullable error))completion {
+  _firestore->WaitForPendingWrites(util::MakeCallback(completion));
+}
+
 - (void)runTransactionWithBlock:(id _Nullable (^)(FIRTransaction *, NSError **))updateBlock
                   dispatchQueue:(dispatch_queue_t)queue
                      completion:

Firestore/Source/Core/FSTFirestoreClient.h

@@ -111,6 +111,13 @@ NS_ASSUME_NONNULL_BEGIN
 - (void)writeMutations:(std::vector<FSTMutation *> &&)mutations
               callback:(util::StatusCallback)callback;
 
+/**
+ * Passes a callback that is triggered when all the pending writes at the
+ * time when this method is called received server acknowledgement.
+ * An acknowledgement can be either acceptance or rejections.
+ */
+- (void)waitForPendingWritesWithCallback:(util::StatusCallback)callback;
+
 /** Tries to execute the transaction in updateCallback up to retries times. */
 - (void)transactionWithRetries:(int)retries
                 updateCallback:(core::TransactionUpdateCallback)updateCallback

Firestore/Source/Core/FSTFirestoreClient.mm

@@ -440,6 +440,20 @@ - (void)writeMutations:(std::vector<FSTMutation *> &&)mutations
   });
 };
 
+- (void)waitForPendingWritesWithCallback:(util::StatusCallback)callback {
+  [self verifyNotShutdown];
+  // Dispatch the result back onto the user dispatch queue.
+  auto async_callback = [self, callback](util::Status status) {
+    if (callback) {
+      self->_userExecutor->Execute([=] { callback(std::move(status)); });
+    }
+  };
+
+  _workerQueue->Enqueue([self, async_callback]() {
+    [self.syncEngine registerPendingWritesCallback:std::move(async_callback)];
+  });
+}
+
 - (void)transactionWithRetries:(int)retries
                 updateCallback:(core::TransactionUpdateCallback)update_callback
                 resultCallback:(core::TransactionResultCallback)resultCallback {

Firestore/Source/Core/FSTSyncEngine.h

@@ -87,6 +87,12 @@ NS_ASSUME_NONNULL_BEGIN
 - (void)writeMutations:(std::vector<FSTMutation *> &&)mutations
             completion:(FSTVoidErrorBlock)completion;
 
+/**
+ * Registers a user callback that is called when all pending mutations at the moment of calling
+ * are acknowledged .
+ */
+- (void)registerPendingWritesCallback:(util::StatusCallback)callback;
+
 /**
  * Runs the given transaction block up to retries times and then calls completion.
  *