Use an explicit ParameterLookup class for better typing.

Author: rkistner

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

@@ -14,15 +14,14 @@ import {
   InternalOpId,
   internalToExternalOpId,
   ProtocolOpId,
-  getLookupBucketDefinitionName,
   ReplicationCheckpoint,
   storage,
   utils,
   WatchWriteCheckpointOptions,
   CHECKPOINT_INVALIDATE_ALL,
   deserializeParameterLookup
 } from '@powersync/service-core';
-import { SqliteJsonRow, SqliteJsonValue, SqlSyncRules } from '@powersync/service-sync-rules';
+import { SqliteJsonRow, ParameterLookup, SqlSyncRules } from '@powersync/service-sync-rules';
 import * as bson from 'bson';
 import { wrapWithAbort } from 'ix/asynciterable/operators/withabort.js';
 import { LRUCache } from 'lru-cache';
@@ -158,7 +157,7 @@ export class MongoSyncBucketStorage
 
     await callback(batch);
     await batch.flush();
-    if (batch.last_flushed_op) {
+    if (batch.last_flushed_op != null) {
       return { flushed_op: batch.last_flushed_op };
     } else {
       return null;
@@ -256,7 +255,7 @@ export class MongoSyncBucketStorage
     return result!;
   }
 
-  async getParameterSets(checkpoint: utils.InternalOpId, lookups: SqliteJsonValue[][]): Promise<SqliteJsonRow[]> {
+  async getParameterSets(checkpoint: utils.InternalOpId, lookups: ParameterLookup[]): Promise<SqliteJsonRow[]> {
     const lookupFilter = lookups.map((lookup) => {
       return storage.serializeLookup(lookup);
     });

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

@@ -1,5 +1,4 @@
 import * as lib_postgres from '@powersync/lib-service-postgres';
-import { ReplicationAssertionError } from '@powersync/lib-services-framework';
 import {
   BroadcastIterable,
   CHECKPOINT_INVALIDATE_ALL,
@@ -14,23 +13,22 @@ import {
 } from '@powersync/service-core';
 import { JSONBig } from '@powersync/service-jsonbig';
 import * as sync_rules from '@powersync/service-sync-rules';
+import * as timers from 'timers/promises';
 import * as uuid from 'uuid';
 import { BIGINT_MAX } from '../types/codecs.js';
 import { models, RequiredOperationBatchLimits } from '../types/types.js';
 import { replicaIdToSubkey } from '../utils/bson.js';
 import { mapOpEntry } from '../utils/bucket-data.js';
-import * as timers from 'timers/promises';
 
 import * as framework from '@powersync/lib-services-framework';
 import { StatementParam } from '@powersync/service-jpgwire';
+import { wrapWithAbort } from 'ix/asynciterable/operators/withabort.js';
 import { SourceTableDecoded, StoredRelationId } from '../types/models/SourceTable.js';
 import { pick } from '../utils/ts-codec.js';
 import { PostgresBucketBatch } from './batch/PostgresBucketBatch.js';
 import { PostgresWriteCheckpointAPI } from './checkpoints/PostgresWriteCheckpointAPI.js';
 import { PostgresBucketStorageFactory } from './PostgresBucketStorageFactory.js';
 import { PostgresCompactor } from './PostgresCompactor.js';
-import { wrapWithAbort } from 'ix/asynciterable/operators/withabort.js';
-import { Decoded } from 'ts-codec';
 
 export type PostgresSyncRulesStorageOptions = {
   factory: PostgresBucketStorageFactory;
@@ -354,7 +352,7 @@ export class PostgresSyncRulesStorage
 
   async getParameterSets(
     checkpoint: utils.InternalOpId,
-    lookups: sync_rules.SqliteJsonValue[][]
+    lookups: sync_rules.ParameterLookup[]
   ): Promise<sync_rules.SqliteJsonRow[]> {
     const rows = await this.db.sql`
       SELECT DISTINCT

packages/service-core-tests/src/tests/register-data-storage-tests.ts

@@ -1,5 +1,5 @@
 import { getUuidReplicaIdentityBson, OplogEntry, storage } from '@powersync/service-core';
-import { RequestParameters } from '@powersync/service-sync-rules';
+import { ParameterLookup, RequestParameters } from '@powersync/service-sync-rules';
 import { expect, test } from 'vitest';
 import * as test_utils from '../test-utils/test-utils-index.js';
 
@@ -65,7 +65,9 @@ bucket_definitions:
       });
     });
 
-    const parameters = await bucketStorage.getParameterSets(result!.flushed_op, [['mybucket', '1', 'user1']]);
+    const parameters = await bucketStorage.getParameterSets(result!.flushed_op, [
+      ParameterLookup.normalized('mybucket', '1', ['user1'])
+    ]);
     expect(parameters).toEqual([
       {
         group_id: 'group1a'
@@ -110,15 +112,19 @@ bucket_definitions:
       });
     });
 
-    const parameters = await bucketStorage.getParameterSets(result2!.flushed_op, [['mybucket', '1', 'user1']]);
+    const parameters = await bucketStorage.getParameterSets(result2!.flushed_op, [
+      ParameterLookup.normalized('mybucket', '1', ['user1'])
+    ]);
     expect(parameters).toEqual([
       {
         group_id: 'group2'
       }
     ]);
 
     // Use the checkpoint to get older data if relevant
-    const parameters2 = await bucketStorage.getParameterSets(result1!.flushed_op, [['mybucket', '1', 'user1']]);
+    const parameters2 = await bucketStorage.getParameterSets(result1!.flushed_op, [
+      ParameterLookup.normalized('mybucket', '1', ['user1'])
+    ]);
     expect(parameters2).toEqual([
       {
         group_id: 'group1'
@@ -183,8 +189,8 @@ bucket_definitions:
     // There removal operation for the association of `list2`::`todo2` should not interfere with the new
     // association of `list1`::`todo2`
     const parameters = await bucketStorage.getParameterSets(result2!.flushed_op, [
-      ['mybucket', '1', 'list1'],
-      ['mybucket', '1', 'list2']
+      ParameterLookup.normalized('mybucket', '1', ['list1']),
+      ParameterLookup.normalized('mybucket', '1', ['list2'])
     ]);
 
     expect(parameters.sort((a, b) => (a.todo_id as string).localeCompare(b.todo_id as string))).toEqual([
@@ -230,11 +236,17 @@ bucket_definitions:
 
     const checkpoint = result!.flushed_op;
 
-    const parameters1 = await bucketStorage.getParameterSets(checkpoint, [['mybucket', '1', 314n, 314, 3.14]]);
+    const parameters1 = await bucketStorage.getParameterSets(checkpoint, [
+      ParameterLookup.normalized('mybucket', '1', [314n, 314, 3.14])
+    ]);
     expect(parameters1).toEqual([TEST_PARAMS]);
-    const parameters2 = await bucketStorage.getParameterSets(checkpoint, [['mybucket', '1', 314, 314n, 3.14]]);
+    const parameters2 = await bucketStorage.getParameterSets(checkpoint, [
+      ParameterLookup.normalized('mybucket', '1', [314, 314n, 3.14])
+    ]);
     expect(parameters2).toEqual([TEST_PARAMS]);
-    const parameters3 = await bucketStorage.getParameterSets(checkpoint, [['mybucket', '1', 314n, 314, 3]]);
+    const parameters3 = await bucketStorage.getParameterSets(checkpoint, [
+      ParameterLookup.normalized('mybucket', '1', [314n, 314, 3])
+    ]);
     expect(parameters3).toEqual([]);
   });
 
@@ -286,7 +298,9 @@ bucket_definitions:
 
     const checkpoint = result!.flushed_op;
 
-    const parameters1 = await bucketStorage.getParameterSets(checkpoint, [['mybucket', '1', 1152921504606846976n]]);
+    const parameters1 = await bucketStorage.getParameterSets(checkpoint, [
+      ParameterLookup.normalized('mybucket', '1', [1152921504606846976n])
+    ]);
     expect(parameters1).toEqual([TEST_PARAMS]);
   });
 
@@ -387,7 +401,7 @@ bucket_definitions:
     const q1 = sync_rules.bucket_descriptors[0].parameter_queries[0];
 
     const lookups = q1.getLookups(parameters);
-    expect(lookups).toEqual([['by_workspace', '1', 'u1']]);
+    expect(lookups).toEqual([ParameterLookup.normalized('by_workspace', '1', ['u1'])]);
 
     const parameter_sets = await bucketStorage.getParameterSets(checkpoint, lookups);
     expect(parameter_sets).toEqual([{ workspace_id: 'workspace1' }]);
@@ -457,7 +471,7 @@ bucket_definitions:
     const q1 = sync_rules.bucket_descriptors[0].parameter_queries[0];
 
     const lookups = q1.getLookups(parameters);
-    expect(lookups).toEqual([['by_public_workspace', '1']]);
+    expect(lookups).toEqual([ParameterLookup.normalized('by_public_workspace', '1', [])]);
 
     const parameter_sets = await bucketStorage.getParameterSets(checkpoint, lookups);
     parameter_sets.sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b)));
@@ -546,15 +560,15 @@ bucket_definitions:
     // Test intermediate values - could be moved to sync_rules.test.ts
     const q1 = sync_rules.bucket_descriptors[0].parameter_queries[0];
     const lookups1 = q1.getLookups(parameters);
-    expect(lookups1).toEqual([['by_workspace', '1']]);
+    expect(lookups1).toEqual([ParameterLookup.normalized('by_workspace', '1', [])]);
 
     const parameter_sets1 = await bucketStorage.getParameterSets(checkpoint, lookups1);
     parameter_sets1.sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b)));
     expect(parameter_sets1).toEqual([{ workspace_id: 'workspace1' }]);
 
     const q2 = sync_rules.bucket_descriptors[0].parameter_queries[1];
     const lookups2 = q2.getLookups(parameters);
-    expect(lookups2).toEqual([['by_workspace', '2', 'u1']]);
+    expect(lookups2).toEqual([ParameterLookup.normalized('by_workspace', '2', ['u1'])]);
 
     const parameter_sets2 = await bucketStorage.getParameterSets(checkpoint, lookups2);
     parameter_sets2.sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b)));
@@ -861,7 +875,9 @@ bucket_definitions:
 
     const { checkpoint } = await bucketStorage.getCheckpoint();
 
-    const parameters = await bucketStorage.getParameterSets(checkpoint, [['mybucket', '1', 'user1']]);
+    const parameters = await bucketStorage.getParameterSets(checkpoint, [
+      ParameterLookup.normalized('mybucket', '1', ['user1'])
+    ]);
     expect(parameters).toEqual([]);
   });
 

packages/service-core/src/storage/SyncRulesBucketStorage.ts

@@ -1,5 +1,5 @@
 import { ObserverClient } from '@powersync/lib-services-framework';
-import { SqlSyncRules, SqliteJsonRow, SqliteJsonValue } from '@powersync/service-sync-rules';
+import { ParameterLookup, SqlSyncRules, SqliteJsonRow } from '@powersync/service-sync-rules';
 import * as util from '../util/util-index.js';
 import { BucketStorageBatch, FlushedResult } from './BucketStorageBatch.js';
 import { BucketStorageFactory } from './BucketStorageFactory.js';
@@ -71,7 +71,7 @@ export interface SyncRulesBucketStorage
   /**
    * Used to resolve "dynamic" parameter queries.
    */
-  getParameterSets(checkpoint: util.InternalOpId, lookups: SqliteJsonValue[][]): Promise<SqliteJsonRow[]>;
+  getParameterSets(checkpoint: util.InternalOpId, lookups: ParameterLookup[]): Promise<SqliteJsonRow[]>;
 
   getCheckpointChanges(options: GetCheckpointChangesOptions): Promise<CheckpointChanges>;
 

packages/service-core/src/storage/bson.ts

@@ -1,6 +1,6 @@
 import * as bson from 'bson';
 
-import { SqliteJsonValue } from '@powersync/service-sync-rules';
+import { ParameterLookup, SqliteJsonValue } from '@powersync/service-sync-rules';
 import { ReplicaId } from './BucketStorageBatch.js';
 
 type NodeBuffer = Buffer<ArrayBuffer>;
@@ -24,11 +24,11 @@ export const BSON_DESERIALIZE_DATA_OPTIONS: bson.DeserializeOptions = {
  * Lookup serialization must be number-agnostic. I.e. normalize numbers, instead of preserving numbers.
  * @param lookup
  */
-export const serializeLookupBuffer = (lookup: SqliteJsonValue[]): NodeBuffer => {
-  return bson.serialize({ l: lookup }) as NodeBuffer;
+export const serializeLookupBuffer = (lookup: ParameterLookup): NodeBuffer => {
+  return bson.serialize({ l: lookup.values }) as NodeBuffer;
 };
 
-export const serializeLookup = (lookup: SqliteJsonValue[]) => {
+export const serializeLookup = (lookup: ParameterLookup) => {
   return new bson.Binary(serializeLookupBuffer(lookup));
 };
 

packages/service-core/src/sync/BucketChecksumState.ts

@@ -288,7 +288,7 @@ export class BucketParameterState {
 
     this.querier = syncRules.getBucketParameterQuerier(this.syncParams);
     this.staticBuckets = new Map<string, BucketDescription>(this.querier.staticBuckets.map((b) => [b.bucket, b]));
-    this.lookups = new Set<string>(this.querier.parameterQueryLookups.map((l) => JSONBig.stringify(l)));
+    this.lookups = new Set<string>(this.querier.parameterQueryLookups.map((l) => JSONBig.stringify(l.values)));
   }
 
   async getCheckpointUpdate(checkpoint: storage.StorageCheckpointUpdate): Promise<CheckpointUpdate | null> {

packages/service-core/test/src/sync/BucketChecksumState.test.ts

@@ -9,7 +9,7 @@ import {
   WatchFilterEvent
 } from '@/index.js';
 import { JSONBig } from '@powersync/service-jsonbig';
-import { RequestParameters, SqliteJsonRow, SqliteJsonValue, SqlSyncRules } from '@powersync/service-sync-rules';
+import { RequestParameters, SqliteJsonRow, ParameterLookup, SqlSyncRules } from '@powersync/service-sync-rules';
 import { describe, expect, test } from 'vitest';
 
 describe('BucketChecksumState', () => {
@@ -475,10 +475,10 @@ bucket_definitions:
 
     storage.getParameterSets = async (
       checkpoint: InternalOpId,
-      lookups: SqliteJsonValue[][]
+      lookups: ParameterLookup[]
     ): Promise<SqliteJsonRow[]> => {
       expect(checkpoint).toEqual(1n);
-      expect(lookups).toEqual([['by_project', '1', 'u1']]);
+      expect(lookups).toEqual([ParameterLookup.normalized('by_project', '1', ['u1'])]);
       return [{ id: 1 }, { id: 2 }];
     };
 
@@ -520,10 +520,10 @@ bucket_definitions:
 
     storage.getParameterSets = async (
       checkpoint: InternalOpId,
-      lookups: SqliteJsonValue[][]
+      lookups: ParameterLookup[]
     ): Promise<SqliteJsonRow[]> => {
       expect(checkpoint).toEqual(2n);
-      expect(lookups).toEqual([['by_project', '1', 'u1']]);
+      expect(lookups).toEqual([ParameterLookup.normalized('by_project', '1', ['u1'])]);
       return [{ id: 1 }, { id: 2 }, { id: 3 }];
     };
 
@@ -581,7 +581,7 @@ class MockBucketChecksumStateStorage implements BucketChecksumStateStorage {
     );
   }
 
-  async getParameterSets(checkpoint: InternalOpId, lookups: SqliteJsonValue[][]): Promise<SqliteJsonRow[]> {
+  async getParameterSets(checkpoint: InternalOpId, lookups: ParameterLookup[]): Promise<SqliteJsonRow[]> {
     throw new Error('Method not implemented.');
   }
 }

packages/sync-rules/src/BucketParameterQuerier.ts

@@ -1,5 +1,6 @@
 import { BucketDescription } from './BucketDescription.js';
 import { RequestParameters, SqliteJsonRow, SqliteJsonValue } from './types.js';
+import { normalizeParameterValue } from './utils.js';
 
 /**
  * Represents a set of parameter queries for a specific request.
@@ -23,7 +24,7 @@ export interface BucketParameterQuerier {
    */
   readonly hasDynamicBuckets: boolean;
 
-  readonly parameterQueryLookups: SqliteJsonValue[][];
+  readonly parameterQueryLookups: ParameterLookup[];
 
   /**
    * These buckets depend on parameter storage, and needs to be retrieved dynamically for each checkpoint.
@@ -39,7 +40,7 @@ export interface BucketParameterQuerier {
 }
 
 export interface ParameterLookupSource {
-  getParameterSets: (lookups: SqliteJsonValue[][]) => Promise<SqliteJsonRow[]>;
+  getParameterSets: (lookups: ParameterLookup[]) => Promise<SqliteJsonRow[]>;
 }
 
 export interface QueryBucketDescriptorOptions extends ParameterLookupSource {
@@ -63,3 +64,25 @@ export function mergeBucketParameterQueriers(queriers: BucketParameterQuerier[])
     }
   };
 }
+
+/**
+ * Represents an equality filter from a parameter query.
+ *
+ * Other query types are not supported yet.
+ */
+export class ParameterLookup {
+  // bucket definition name, parameter query index, ...lookup values
+  readonly values: SqliteJsonValue[];
+
+  static normalized(bucketDefinitionName: string, queryIndex: string, values: SqliteJsonValue[]): ParameterLookup {
+    return new ParameterLookup([bucketDefinitionName, queryIndex, ...values.map(normalizeParameterValue)]);
+  }
+
+  /**
+   *
+   * @param values must be pre-normalized (any integer converted into bigint)
+   */
+  constructor(values: SqliteJsonValue[]) {
+    this.values = values;
+  }
+}