Workaround for booleans; log the filters.

Author: rkistner

modules/module-mongodb/src/replication/MongoSnapshotQuery.ts

@@ -1,8 +1,7 @@
 import { mongo } from '@powersync/lib-service-mongodb';
 import { ReplicationAssertionError } from '@powersync/lib-services-framework';
 import { bson } from '@powersync/service-core';
-import { MongoExpression, StaticFilter } from '@powersync/service-sync-rules';
-import { staticFilterToMongoExpression } from './staticFilters.js';
+import { MongoExpression } from '@powersync/service-sync-rules';
 
 /**
  * Performs a collection snapshot query, chunking by ranges of _id.
@@ -21,15 +20,13 @@ export class ChunkedSnapshotQuery implements AsyncDisposable {
     collection: mongo.Collection;
     batchSize: number;
     key?: Uint8Array | null;
-    filter?: StaticFilter;
+    filter?: MongoExpression;
   }) {
     this.lastKey = options.key ? bson.deserialize(options.key, { useBigInt64: true })._id : null;
     this.lastCursor = null;
     this.collection = options.collection;
     this.batchSize = options.batchSize;
-    if (options.filter) {
-      this.filter = staticFilterToMongoExpression(options.filter);
-    }
+    this.filter = options.filter ?? null;
   }
 
   async nextChunk(): Promise<{ docs: mongo.Document[]; lastKey: Uint8Array } | { docs: []; lastKey: null }> {

modules/module-mongodb/src/replication/MongoSnapshotter.ts

@@ -20,6 +20,8 @@ import { MongoManager } from './MongoManager.js';
 import { constructAfterRecord, createCheckpoint, getMongoRelation, STANDALONE_CHECKPOINT_ID } from './MongoRelation.js';
 import { ChunkedSnapshotQuery } from './MongoSnapshotQuery.js';
 import { CHECKPOINTS_COLLECTION } from './replication-utils.js';
+import { staticFilterToMongoExpression } from './staticFilters.js';
+import { JSONBig } from '@powersync/service-jsonbig';
 
 export interface MongoSnapshotterOptions {
   connections: MongoManager;
@@ -281,19 +283,22 @@ export class MongoSnapshotter {
     let at = table.snapshotStatus?.replicatedCount ?? 0;
     const db = this.client.db(table.schema);
     const collection = db.collection(table.name);
-    console.log('snapshot with filter', table.pattern?.filter, table.pattern);
+
+    const mongoFilter = table.pattern?.filter ? staticFilterToMongoExpression(table.pattern.filter) : null;
+    const filterLog = mongoFilter ? ` | filter: ${JSONBig.stringify(mongoFilter)}` : '';
+
     await using query = new ChunkedSnapshotQuery({
       collection,
       key: table.snapshotStatus?.lastKey,
       batchSize: this.snapshotChunkLength,
-      filter: table.pattern?.filter
+      filter: mongoFilter
     });
     if (query.lastKey != null) {
       this.logger.info(
-        `Replicating ${table.qualifiedName} ${table.formatSnapshotProgress()} - resuming at _id > ${query.lastKey}`
+        `Replicating ${table.qualifiedName} ${table.formatSnapshotProgress()} - resuming at _id > ${query.lastKey}${filterLog}`
       );
     } else {
-      this.logger.info(`Replicating ${table.qualifiedName} ${table.formatSnapshotProgress()}`);
+      this.logger.info(`Replicating ${table.qualifiedName} ${table.formatSnapshotProgress()}${filterLog}`);
     }
 
     let lastBatch = performance.now();
@@ -355,21 +360,6 @@ export class MongoSnapshotter {
     return rowProcessor.applyRowContext<never>(inputRow);
   }
 
-  private async getCollectionInfo(db: string, name: string): Promise<mongo.CollectionInfo | undefined> {
-    const collection = (
-      await this.client
-        .db(db)
-        .listCollections(
-          {
-            name: name
-          },
-          { nameOnly: false }
-        )
-        .toArray()
-    )[0];
-    return collection;
-  }
-
   private async checkPostImages(db: string, collectionInfo: mongo.CollectionInfo) {
     if (!this.usePostImages) {
       // Nothing to check

modules/module-mongodb/src/replication/staticFilters.ts

@@ -72,9 +72,23 @@ export function staticFilterToMongoExpression(
       switch (operator) {
         case '=':
         case '==':
+        case 'IS':
+          // Big hack to support boolean values coerced to integers.
+          if ('value' in node.right && node.right.value == 0n) {
+            return { $in: [leftExpr, [0n, false]] };
+          } else if ('value' in node.right && node.right.value == 1n) {
+            return { $in: [leftExpr, [1n, true]] };
+          }
           return { $eq: [leftExpr, rightExpr] };
         case '!=':
         case '<>':
+        case 'IS NOT':
+          // Big hack to support boolean values coerced to integers.
+          if ('value' in node.right && node.right.value == 0n) {
+            return { $nin: [leftExpr, [0n, false]] };
+          } else if ('value' in node.right && node.right.value == 1n) {
+            return { $nin: [leftExpr, [1n, true]] };
+          }
           return { $ne: [leftExpr, rightExpr] };
         case '<':
           return { $lt: [leftExpr, rightExpr] };
@@ -84,10 +98,6 @@ export function staticFilterToMongoExpression(
           return { $gt: [leftExpr, rightExpr] };
         case '>=':
           return { $gte: [leftExpr, rightExpr] };
-        case 'IS':
-          return { $eq: [leftExpr, rightExpr] };
-        case 'IS NOT':
-          return { $ne: [leftExpr, rightExpr] };
         case 'IN': {
           if (parseJsonArrayForIn && 'value' in node.right) {
             const value = node.right.value;
@@ -108,7 +118,7 @@ export function staticFilterToMongoExpression(
           return { $in: [leftExpr, rightExpr] };
         }
         default:
-          throw new Error(`Operator not supported: ${node.operator}`);
+          return ANY;
       }
     }
 

packages/service-core-tests/src/test-utils/general-utils.ts

@@ -143,7 +143,6 @@ export function bucketRequest(
     throw new Error('Failed to find global bucket');
   }
   const bucketName = hydrationState.getBucketSourceScope(source).bucketPrefix + parameters;
-  console.log('query for bucket', bucketName);
   return {
     bucket: bucketName,
     start: BigInt(start ?? 0n),