Refactor getParameterSets to take a ReplicationCheckpoint.

Author: rkistner

modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts

@@ -261,7 +261,7 @@ export class MongoSyncBucketStorage
     return result!;
   }
 
-  async getParameterSets(checkpoint: utils.InternalOpId, lookups: ParameterLookup[]): Promise<SqliteJsonRow[]> {
+  async getParameterSets(checkpoint: ReplicationCheckpoint, lookups: ParameterLookup[]): Promise<SqliteJsonRow[]> {
     const lookupFilter = lookups.map((lookup) => {
       return storage.serializeLookup(lookup);
     });
@@ -271,7 +271,7 @@ export class MongoSyncBucketStorage
           $match: {
             'key.g': this.group_id,
             lookup: { $in: lookupFilter },
-            _id: { $lte: checkpoint }
+            _id: { $lte: checkpoint.checkpoint }
           }
         },
         {

modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts

@@ -8,6 +8,7 @@ import {
   internalToExternalOpId,
   LastValueSink,
   maxLsn,
+  ReplicationCheckpoint,
   storage,
   utils,
   WatchWriteCheckpointOptions
@@ -350,7 +351,7 @@ export class PostgresSyncRulesStorage
   }
 
   async getParameterSets(
-    checkpoint: utils.InternalOpId,
+    checkpoint: ReplicationCheckpoint,
     lookups: sync_rules.ParameterLookup[]
   ): Promise<sync_rules.SqliteJsonRow[]> {
     const rows = await this.db.sql`
@@ -373,7 +374,7 @@ export class PostgresSyncRulesStorage
         value: lookups.map((l) => storage.serializeLookupBuffer(l).toString('hex'))
       }}) AS FILTER
         )
-        AND id <= ${{ type: 'int8', value: checkpoint }}
+        AND id <= ${{ type: 'int8', value: checkpoint.checkpoint }}
       ORDER BY
         lookup,
         source_table,