Listen sequence numbers (#675)

* Generate and save sequence numbers for listens

* Add documentation

* Fix include path

* Fix unavailable comment

* Review feedback

7 files changed, 50 insertions, 3 deletions

diff --git a/Firestore/Source/Local/FSTLevelDBQueryCache.mm b/Firestore/Source/Local/FSTLevelDBQueryCache.mm
index 13d15ee..8388b96 100644
--- a/Firestore/Source/Local/FSTLevelDBQueryCache.mm
+++ b/Firestore/Source/Local/FSTLevelDBQueryCache.mm
@@ -100,6 +100,10 @@ static ReadOptions GetStandardReadOptions() {
   return self.metadata.highestTargetId;
 }
 
+- (FSTListenSequenceNumber)highestListenSequenceNumber {
+  return self.metadata.highestListenSequenceNumber;
+}
+
 - (FSTSnapshotVersion *)lastRemoteSnapshotVersion {
   return _lastRemoteSnapshotVersion;
 }
@@ -116,7 +120,6 @@ static ReadOptions GetStandardReadOptions() {
 }
 
 - (void)addQueryData:(FSTQueryData *)queryData group:(FSTWriteGroup *)group {
-  // TODO(mcg): actually populate listen sequence number
   FSTTargetID targetID = queryData.targetID;
   std::string key = [FSTLevelDBTargetKey keyWithTargetID:targetID];
   [group setMessage:[self.serializer encodedQueryData:queryData] forKey:key];
@@ -127,9 +130,19 @@ static ReadOptions GetStandardReadOptions() {
   std::string emptyBuffer;
   [group setData:emptyBuffer forKey:indexKey];
 
+  BOOL saveMetadata = NO;
   FSTPBTargetGlobal *metadata = self.metadata;
   if (targetID > metadata.highestTargetId) {
     metadata.highestTargetId = targetID;
+    saveMetadata = YES;
+  }
+
+  if (queryData.sequenceNumber > metadata.highestListenSequenceNumber) {
+    metadata.highestListenSequenceNumber = queryData.sequenceNumber;
+    saveMetadata = YES;
+  }
+
+  if (saveMetadata) {
     [group setMessage:metadata forKey:[FSTLevelDBTargetGlobalKey key]];
   }
 }
diff --git a/Firestore/Source/Local/FSTLocalSerializer.m b/Firestore/Source/Local/FSTLocalSerializer.m
index c71e9dd..82aec4d 100644
--- a/Firestore/Source/Local/FSTLocalSerializer.m
+++ b/Firestore/Source/Local/FSTLocalSerializer.m
@@ -156,6 +156,7 @@
 
   FSTPBTarget *proto = [FSTPBTarget message];
   proto.targetId = queryData.targetID;
+  proto.lastListenSequenceNumber = queryData.sequenceNumber;
   proto.snapshotVersion = [remoteSerializer encodedVersion:queryData.snapshotVersion];
   proto.resumeToken = queryData.resumeToken;
 
@@ -173,6 +174,7 @@
   FSTSerializerBeta *remoteSerializer = self.remoteSerializer;
 
   FSTTargetID targetID = target.targetId;
+  FSTListenSequenceNumber sequenceNumber = target.lastListenSequenceNumber;
   FSTSnapshotVersion *version = [remoteSerializer decodedVersion:target.snapshotVersion];
   NSData *resumeToken = target.resumeToken;
 
@@ -192,6 +194,7 @@
 
   return [[FSTQueryData alloc] initWithQuery:query
                                     targetID:targetID
+                        listenSequenceNumber:sequenceNumber
                                      purpose:FSTQueryPurposeListen
                              snapshotVersion:version
                                  resumeToken:resumeToken];
diff --git a/Firestore/Source/Local/FSTLocalStore.m b/Firestore/Source/Local/FSTLocalStore.m
index cde7104..3a5b0b4 100644
--- a/Firestore/Source/Local/FSTLocalStore.m
+++ b/Firestore/Source/Local/FSTLocalStore.m
@@ -17,6 +17,7 @@
 #import "Firestore/Source/Local/FSTLocalStore.h"
 
 #import "Firestore/Source/Auth/FSTUser.h"
+#import "Firestore/Source/Core/FSTListenSequence.h"
 #import "Firestore/Source/Core/FSTQuery.h"
 #import "Firestore/Source/Core/FSTSnapshotVersion.h"
 #import "Firestore/Source/Core/FSTTargetIDGenerator.h"
@@ -76,6 +77,8 @@ NS_ASSUME_NONNULL_BEGIN
 /** Used to generate targetIDs for queries tracked locally. */
 @property(nonatomic, strong) FSTTargetIDGenerator *targetIDGenerator;
 
+@property(nonatomic, strong) FSTListenSequence *listenSequence;
+
 /**
  * A heldBatchResult is a mutation batch result (from a write acknowledgement) that arrived before
  * the watch stream got notified of a snapshot that includes the write.  So we "hold" it until
@@ -148,6 +151,8 @@ NS_ASSUME_NONNULL_BEGIN
 
   FSTTargetID targetID = [self.queryCache highestTargetID];
   self.targetIDGenerator = [FSTTargetIDGenerator generatorForLocalStoreStartingAfterID:targetID];
+  FSTListenSequenceNumber sequenceNumber = [self.queryCache highestListenSequenceNumber];
+  self.listenSequence = [[FSTListenSequence alloc] initStartingAfter:sequenceNumber];
 }
 
 - (void)shutdown {
@@ -380,6 +385,7 @@ NS_ASSUME_NONNULL_BEGIN
 - (FSTQueryData *)allocateQuery:(FSTQuery *)query {
   FSTQueryData *cached = [self.queryCache queryDataForQuery:query];
   FSTTargetID targetID;
+  FSTListenSequenceNumber sequenceNumber = [self.listenSequence next];
   if (cached) {
     // This query has been listened to previously, so reuse the previous targetID.
     // TODO(mcg): freshen last accessed date?
@@ -388,8 +394,10 @@ NS_ASSUME_NONNULL_BEGIN
     FSTWriteGroup *group = [self.persistence startGroupWithAction:@"Allocate query"];
 
     targetID = [self.targetIDGenerator nextID];
-    cached =
-        [[FSTQueryData alloc] initWithQuery:query targetID:targetID purpose:FSTQueryPurposeListen];
+    cached = [[FSTQueryData alloc] initWithQuery:query
+                                        targetID:targetID
+                            listenSequenceNumber:sequenceNumber
+                                         purpose:FSTQueryPurposeListen];
     [self.queryCache addQueryData:cached group:group];
 
     [self.persistence commitGroup:group];
diff --git a/Firestore/Source/Local/FSTMemoryQueryCache.m b/Firestore/Source/Local/FSTMemoryQueryCache.m
index 8d37bcb..bcab174 100644
--- a/Firestore/Source/Local/FSTMemoryQueryCache.m
+++ b/Firestore/Source/Local/FSTMemoryQueryCache.m
@@ -34,6 +34,8 @@ NS_ASSUME_NONNULL_BEGIN
 /** The highest numbered target ID encountered. */
 @property(nonatomic, assign) FSTTargetID highestTargetID;
 
+@property(nonatomic, assign) FSTListenSequenceNumber highestListenSequenceNumber;
+
 @end
 
 @implementation FSTMemoryQueryCache {
@@ -65,6 +67,10 @@ NS_ASSUME_NONNULL_BEGIN
   return _highestTargetID;
 }
 
+- (FSTListenSequenceNumber)highestListenSequenceNumber {
+  return _highestListenSequenceNumber;
+}
+
 - (FSTSnapshotVersion *)lastRemoteSnapshotVersion {
   return _lastRemoteSnapshotVersion;
 }
@@ -79,6 +85,9 @@ NS_ASSUME_NONNULL_BEGIN
   if (queryData.targetID > self.highestTargetID) {
     self.highestTargetID = queryData.targetID;
   }
+  if (queryData.sequenceNumber > self.highestListenSequenceNumber) {
+    self.highestListenSequenceNumber = queryData.sequenceNumber;
+  }
 }
 
 - (void)removeQueryData:(FSTQueryData *)queryData group:(__unused FSTWriteGroup *)group {
diff --git a/Firestore/Source/Local/FSTQueryCache.h b/Firestore/Source/Local/FSTQueryCache.h
index e0cf4c8..88c9df9 100644
--- a/Firestore/Source/Local/FSTQueryCache.h
+++ b/Firestore/Source/Local/FSTQueryCache.h
@@ -53,6 +53,11 @@ NS_ASSUME_NONNULL_BEGIN
 - (FSTTargetID)highestTargetID;
 
 /**
+ * Returns the highest listen sequence number of any query seen by the cache.
+ */
+- (FSTListenSequenceNumber)highestListenSequenceNumber;
+
+/**
  * A global snapshot version representing the last consistent snapshot we received from the
  * backend. This is monotonically increasing and any snapshots received from the backend prior to
  * this version (e.g. for targets resumed with a resume_token) should be suppressed (buffered)
diff --git a/Firestore/Source/Local/FSTQueryData.h b/Firestore/Source/Local/FSTQueryData.h
index 048bfad..5db2de6 100644
--- a/Firestore/Source/Local/FSTQueryData.h
+++ b/Firestore/Source/Local/FSTQueryData.h
@@ -40,6 +40,7 @@ typedef NS_ENUM(NSInteger, FSTQueryPurpose) {
 
 - (instancetype)initWithQuery:(FSTQuery *)query
                      targetID:(FSTTargetID)targetID
+         listenSequenceNumber:(FSTListenSequenceNumber)sequenceNumber
                       purpose:(FSTQueryPurpose)purpose
               snapshotVersion:(FSTSnapshotVersion *)snapshotVersion
                   resumeToken:(NSData *)resumeToken NS_DESIGNATED_INITIALIZER;
@@ -47,6 +48,7 @@ typedef NS_ENUM(NSInteger, FSTQueryPurpose) {
 /** Convenience initializer for use when creating an FSTQueryData for the first time. */
 - (instancetype)initWithQuery:(FSTQuery *)query
                      targetID:(FSTTargetID)targetID
+         listenSequenceNumber:(FSTListenSequenceNumber)sequenceNumber
                       purpose:(FSTQueryPurpose)purpose;
 
 - (instancetype)init NS_UNAVAILABLE;
@@ -64,6 +66,8 @@ typedef NS_ENUM(NSInteger, FSTQueryPurpose) {
  */
 @property(nonatomic, assign, readonly) FSTTargetID targetID;
 
+@property(nonatomic, assign, readonly) FSTListenSequenceNumber sequenceNumber;
+
 /** The purpose of the query. */
 @property(nonatomic, assign, readonly) FSTQueryPurpose purpose;
 
diff --git a/Firestore/Source/Local/FSTQueryData.m b/Firestore/Source/Local/FSTQueryData.m
index 080f136..6bb716a 100644
--- a/Firestore/Source/Local/FSTQueryData.m
+++ b/Firestore/Source/Local/FSTQueryData.m
@@ -25,6 +25,7 @@ NS_ASSUME_NONNULL_BEGIN
 
 - (instancetype)initWithQuery:(FSTQuery *)query
                      targetID:(FSTTargetID)targetID
+         listenSequenceNumber:(FSTListenSequenceNumber)sequenceNumber
                       purpose:(FSTQueryPurpose)purpose
               snapshotVersion:(FSTSnapshotVersion *)snapshotVersion
                   resumeToken:(NSData *)resumeToken {
@@ -32,6 +33,7 @@ NS_ASSUME_NONNULL_BEGIN
   if (self) {
     _query = query;
     _targetID = targetID;
+    _sequenceNumber = sequenceNumber;
     _purpose = purpose;
     _snapshotVersion = snapshotVersion;
     _resumeToken = [resumeToken copy];
@@ -41,9 +43,11 @@ NS_ASSUME_NONNULL_BEGIN
 
 - (instancetype)initWithQuery:(FSTQuery *)query
                      targetID:(FSTTargetID)targetID
+         listenSequenceNumber:(FSTListenSequenceNumber)sequenceNumber
                       purpose:(FSTQueryPurpose)purpose {
   return [self initWithQuery:query
                     targetID:targetID
+        listenSequenceNumber:sequenceNumber
                      purpose:purpose
              snapshotVersion:[FSTSnapshotVersion noVersion]
                  resumeToken:[NSData data]];
@@ -83,6 +87,7 @@ NS_ASSUME_NONNULL_BEGIN
                                         resumeToken:(NSData *)resumeToken {
   return [[FSTQueryData alloc] initWithQuery:self.query
                                     targetID:self.targetID
+                        listenSequenceNumber:self.sequenceNumber
                                      purpose:self.purpose
                              snapshotVersion:snapshotVersion
                                  resumeToken:resumeToken];