Validate changeStreamPreAndPostImages on existing collections.

rkistner · rkistner · commit c5ca6e728e1f · 2024-11-13T12:45:40.000+02:00
diff --git a/modules/module-mongodb/src/api/MongoRouteAPIAdapter.ts b/modules/module-mongodb/src/api/MongoRouteAPIAdapter.ts
@@ -34,6 +34,10 @@ export class MongoRouteAPIAdapter implements api.RouteAPI {
     await this.client.close();
   }
 
+  async [Symbol.asyncDispose]() {
+    await this.shutdown();
+  }
+
   async getSourceConfig(): Promise<service_types.configFile.ResolvedDataSourceConfig> {
     return this.config;
   }
@@ -77,6 +81,28 @@ export class MongoRouteAPIAdapter implements api.RouteAPI {
     sqlSyncRules: sync_rules.SqlSyncRules
   ): Promise<api.PatternResult[]> {
     let result: api.PatternResult[] = [];
+
+    const validatePostImages = (schema: string, collection: mongo.CollectionInfo): service_types.ReplicationError[] => {
+      if (this.config.postImages == types.PostImagesOption.OFF) {
+        return [];
+      } else if (!collection.options?.changeStreamPreAndPostImages?.enabled) {
+        if (this.config.postImages == types.PostImagesOption.READ_ONLY) {
+          return [
+            { level: 'fatal', message: `changeStreamPreAndPostImages not enabled on ${schema}.${collection.name}` }
+          ];
+        } else {
+          return [
+            {
+              level: 'warning',
+              message: `changeStreamPreAndPostImages not enabled on ${schema}.${collection.name}, will be enabled automatically`
+            }
+          ];
+        }
+      } else {
+        return [];
+      }
+    };
+
     for (let tablePattern of tablePatterns) {
       const schema = tablePattern.schema;
 
@@ -101,7 +127,7 @@ export class MongoRouteAPIAdapter implements api.RouteAPI {
           {
             name: nameFilter
           },
-          { nameOnly: true }
+          { nameOnly: false }
         )
         .toArray();
 
@@ -117,6 +143,12 @@ export class MongoRouteAPIAdapter implements api.RouteAPI {
             [],
             true
           );
+          let errors: service_types.ReplicationError[] = [];
+          if (collection.type == 'view') {
+            errors.push({ level: 'warning', message: `Collection ${schema}.${tablePattern.name} is a view` });
+          } else {
+            errors.push(...validatePostImages(schema, collection));
+          }
           const syncData = sqlSyncRules.tableSyncsData(sourceTable);
           const syncParameters = sqlSyncRules.tableSyncsParameters(sourceTable);
           patternResult.tables.push({
@@ -125,7 +157,7 @@ export class MongoRouteAPIAdapter implements api.RouteAPI {
             replication_id: ['_id'],
             data_queries: syncData,
             parameter_queries: syncParameters,
-            errors: []
+            errors: errors
           });
         }
       } else {
@@ -141,26 +173,25 @@ export class MongoRouteAPIAdapter implements api.RouteAPI {
 
         const syncData = sqlSyncRules.tableSyncsData(sourceTable);
         const syncParameters = sqlSyncRules.tableSyncsParameters(sourceTable);
+        const collection = collections[0];
 
-        if (collections.length == 1) {
-          patternResult.table = {
-            schema,
-            name: tablePattern.name,
-            replication_id: ['_id'],
-            data_queries: syncData,
-            parameter_queries: syncParameters,
-            errors: []
-          };
-        } else {
-          patternResult.table = {
-            schema,
-            name: tablePattern.name,
-            replication_id: ['_id'],
-            data_queries: syncData,
-            parameter_queries: syncParameters,
-            errors: [{ level: 'warning', message: `Collection ${schema}.${tablePattern.name} not found` }]
-          };
+        let errors: service_types.ReplicationError[] = [];
+        if (collections.length != 1) {
+          errors.push({ level: 'warning', message: `Collection ${schema}.${tablePattern.name} not found` });
+        } else if (collection.type == 'view') {
+          errors.push({ level: 'warning', message: `Collection ${schema}.${tablePattern.name} is a view` });
+        } else if (!collection.options?.changeStreamPreAndPostImages?.enabled) {
+          errors.push(...validatePostImages(schema, collection));
         }
+
+        patternResult.table = {
+          schema,
+          name: tablePattern.name,
+          replication_id: ['_id'],
+          data_queries: syncData,
+          parameter_queries: syncParameters,
+          errors
+        };
       }
     }
     return result;
diff --git a/modules/module-mongodb/test/src/mongo_test.test.ts b/modules/module-mongodb/test/src/mongo_test.test.ts
@@ -1,10 +1,11 @@
 import { MongoRouteAPIAdapter } from '@module/api/MongoRouteAPIAdapter.js';
 import { ChangeStream } from '@module/replication/ChangeStream.js';
 import { constructAfterRecord } from '@module/replication/MongoRelation.js';
-import { SqliteRow } from '@powersync/service-sync-rules';
+import { SqliteRow, SqlSyncRules } from '@powersync/service-sync-rules';
 import * as mongo from 'mongodb';
 import { describe, expect, test } from 'vitest';
 import { clearTestDb, connectMongoData, TEST_CONNECTION_OPTIONS } from './util.js';
+import { PostImagesOption } from '@module/types/types.js';
 
 describe('mongo data types', () => {
   async function setupTable(db: mongo.Db) {
@@ -245,58 +246,191 @@ describe('mongo data types', () => {
   });
 
   test('connection schema', async () => {
-    const adapter = new MongoRouteAPIAdapter({
+    await using adapter = new MongoRouteAPIAdapter({
       type: 'mongodb',
       ...TEST_CONNECTION_OPTIONS
     });
-    try {
-      const db = adapter.db;
-      await clearTestDb(db);
+    const db = adapter.db;
+    await clearTestDb(db);
 
-      const collection = db.collection('test_data');
-      await setupTable(db);
-      await insert(collection);
+    const collection = db.collection('test_data');
+    await setupTable(db);
+    await insert(collection);
+
+    const schema = await adapter.getConnectionSchema();
+    const dbSchema = schema.filter((s) => s.name == TEST_CONNECTION_OPTIONS.database)[0];
+    expect(dbSchema).not.toBeNull();
+    expect(dbSchema.tables).toMatchObject([
+      {
+        name: 'test_data',
+        columns: [
+          { name: '_id', sqlite_type: 4, internal_type: 'Integer' },
+          { name: 'bool', sqlite_type: 4, internal_type: 'Boolean' },
+          { name: 'bytea', sqlite_type: 1, internal_type: 'Binary' },
+          { name: 'date', sqlite_type: 2, internal_type: 'Date' },
+          { name: 'decimal', sqlite_type: 2, internal_type: 'Decimal' },
+          { name: 'float', sqlite_type: 8, internal_type: 'Double' },
+          { name: 'int2', sqlite_type: 4, internal_type: 'Integer' },
+          { name: 'int4', sqlite_type: 4, internal_type: 'Integer' },
+          { name: 'int8', sqlite_type: 4, internal_type: 'Long' },
+          // We can fix these later
+          { name: 'js', sqlite_type: 2, internal_type: 'Object' },
+          { name: 'js2', sqlite_type: 2, internal_type: 'Object' },
+          { name: 'maxKey', sqlite_type: 0, internal_type: 'MaxKey' },
+          { name: 'minKey', sqlite_type: 0, internal_type: 'MinKey' },
+          { name: 'nested', sqlite_type: 2, internal_type: 'Object' },
+          { name: 'null', sqlite_type: 0, internal_type: 'Null' },
+          { name: 'objectId', sqlite_type: 2, internal_type: 'ObjectId' },
+          // We can fix these later
+          { name: 'pointer', sqlite_type: 2, internal_type: 'Object' },
+          { name: 'pointer2', sqlite_type: 2, internal_type: 'Object' },
+          { name: 'regexp', sqlite_type: 2, internal_type: 'RegExp' },
+          // Can fix this later
+          { name: 'symbol', sqlite_type: 2, internal_type: 'String' },
+          { name: 'text', sqlite_type: 2, internal_type: 'String' },
+          { name: 'timestamp', sqlite_type: 4, internal_type: 'Timestamp' },
+          { name: 'undefined', sqlite_type: 0, internal_type: 'Null' },
+          { name: 'uuid', sqlite_type: 2, internal_type: 'UUID' }
+        ]
+      }
+    ]);
+  });
+
+  test('validate postImages', async () => {
+    await using adapter = new MongoRouteAPIAdapter({
+      type: 'mongodb',
+      ...TEST_CONNECTION_OPTIONS,
+      postImages: PostImagesOption.READ_ONLY
+    });
+    const db = adapter.db;
+    await clearTestDb(db);
+
+    const collection = db.collection('test_data');
+    await setupTable(db);
+    await insert(collection);
+
+    const rules = SqlSyncRules.fromYaml(
+      `
+bucket_definitions:
+  global:
+    data:
+      - select _id as id, * from test_data
 
-      const schema = await adapter.getConnectionSchema();
-      const dbSchema = schema.filter((s) => s.name == TEST_CONNECTION_OPTIONS.database)[0];
-      expect(dbSchema).not.toBeNull();
-      expect(dbSchema.tables).toMatchObject([
+      `,
+      {
+        ...adapter.getParseSyncRulesOptions(),
+        // No schema-based validation at this point
+        schema: undefined
+      }
+    );
+    const source_table_patterns = rules.getSourceTables();
+    const results = await adapter.getDebugTablesInfo(source_table_patterns, rules);
+
+    const result = results[0];
+    expect(result).not.toBeNull();
+    expect(result.table).toMatchObject({
+      schema: 'powersync_test_data',
+      name: 'test_data',
+      replication_id: ['_id'],
+      data_queries: true,
+      parameter_queries: false,
+      errors: [
         {
-          name: 'test_data',
-          columns: [
-            { name: '_id', sqlite_type: 4, internal_type: 'Integer' },
-            { name: 'bool', sqlite_type: 4, internal_type: 'Boolean' },
-            { name: 'bytea', sqlite_type: 1, internal_type: 'Binary' },
-            { name: 'date', sqlite_type: 2, internal_type: 'Date' },
-            { name: 'decimal', sqlite_type: 2, internal_type: 'Decimal' },
-            { name: 'float', sqlite_type: 8, internal_type: 'Double' },
-            { name: 'int2', sqlite_type: 4, internal_type: 'Integer' },
-            { name: 'int4', sqlite_type: 4, internal_type: 'Integer' },
-            { name: 'int8', sqlite_type: 4, internal_type: 'Long' },
-            // We can fix these later
-            { name: 'js', sqlite_type: 2, internal_type: 'Object' },
-            { name: 'js2', sqlite_type: 2, internal_type: 'Object' },
-            { name: 'maxKey', sqlite_type: 0, internal_type: 'MaxKey' },
-            { name: 'minKey', sqlite_type: 0, internal_type: 'MinKey' },
-            { name: 'nested', sqlite_type: 2, internal_type: 'Object' },
-            { name: 'null', sqlite_type: 0, internal_type: 'Null' },
-            { name: 'objectId', sqlite_type: 2, internal_type: 'ObjectId' },
-            // We can fix these later
-            { name: 'pointer', sqlite_type: 2, internal_type: 'Object' },
-            { name: 'pointer2', sqlite_type: 2, internal_type: 'Object' },
-            { name: 'regexp', sqlite_type: 2, internal_type: 'RegExp' },
-            // Can fix this later
-            { name: 'symbol', sqlite_type: 2, internal_type: 'String' },
-            { name: 'text', sqlite_type: 2, internal_type: 'String' },
-            { name: 'timestamp', sqlite_type: 4, internal_type: 'Timestamp' },
-            { name: 'undefined', sqlite_type: 0, internal_type: 'Null' },
-            { name: 'uuid', sqlite_type: 2, internal_type: 'UUID' }
-          ]
+          level: 'fatal',
+          message: 'changeStreamPreAndPostImages not enabled on powersync_test_data.test_data'
         }
-      ]);
-    } finally {
-      await adapter.shutdown();
-    }
+      ]
+    });
+  });
+
+  test('validate postImages - auto-configure', async () => {
+    await using adapter = new MongoRouteAPIAdapter({
+      type: 'mongodb',
+      ...TEST_CONNECTION_OPTIONS,
+      postImages: PostImagesOption.AUTO_CONFIGURE
+    });
+    const db = adapter.db;
+    await clearTestDb(db);
+
+    const collection = db.collection('test_data');
+    await setupTable(db);
+    await insert(collection);
+
+    const rules = SqlSyncRules.fromYaml(
+      `
+bucket_definitions:
+  global:
+    data:
+      - select _id as id, * from test_data
+
+      `,
+      {
+        ...adapter.getParseSyncRulesOptions(),
+        // No schema-based validation at this point
+        schema: undefined
+      }
+    );
+    const source_table_patterns = rules.getSourceTables();
+    const results = await adapter.getDebugTablesInfo(source_table_patterns, rules);
+
+    const result = results[0];
+    expect(result).not.toBeNull();
+    expect(result.table).toMatchObject({
+      schema: 'powersync_test_data',
+      name: 'test_data',
+      replication_id: ['_id'],
+      data_queries: true,
+      parameter_queries: false,
+      errors: [
+        {
+          level: 'warning',
+          message:
+            'changeStreamPreAndPostImages not enabled on powersync_test_data.test_data, will be enabled automatically'
+        }
+      ]
+    });
+  });
+
+  test('validate postImages - off', async () => {
+    await using adapter = new MongoRouteAPIAdapter({
+      type: 'mongodb',
+      ...TEST_CONNECTION_OPTIONS,
+      postImages: PostImagesOption.OFF
+    });
+    const db = adapter.db;
+    await clearTestDb(db);
+
+    const collection = db.collection('test_data');
+    await setupTable(db);
+    await insert(collection);
+
+    const rules = SqlSyncRules.fromYaml(
+      `
+bucket_definitions:
+  global:
+    data:
+      - select _id as id, * from test_data
+
+      `,
+      {
+        ...adapter.getParseSyncRulesOptions(),
+        // No schema-based validation at this point
+        schema: undefined
+      }
+    );
+    const source_table_patterns = rules.getSourceTables();
+    const results = await adapter.getDebugTablesInfo(source_table_patterns, rules);
+
+    const result = results[0];
+    expect(result).not.toBeNull();
+    expect(result.table).toMatchObject({
+      schema: 'powersync_test_data',
+      name: 'test_data',
+      replication_id: ['_id'],
+      data_queries: true,
+      parameter_queries: false,
+      errors: []
+    });
   });
 });
 
