MySQL Alpha Support

.changeset/rotten-pumas-protect.md

@@ -0,0 +1,9 @@
+---
+'@powersync/service-core': minor
+'@powersync/service-module-mysql': minor
+'@powersync/service-sync-rules': minor
+---
+
+Introduced alpha support for MySQL as a datasource for replication.
+Bunch of cleanup
+

.github/workflows/test.yml

@@ -22,7 +22,7 @@ jobs:
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
 
-      - name: Build and push
+      - name: Test Build Docker Image
         uses: docker/build-push-action@v5
         with:
           cache-from: type=registry,ref=stevenontong/${{vars.DOCKER_REGISTRY}}:cache

README.md

@@ -2,7 +2,7 @@
   <a href="https://www.powersync.com" target="_blank"><img src="https://github.com/powersync-ja/.github/assets/7372448/d2538c43-c1a0-4c47-9a76-41462dba484f"/></a>
 </p>
 
-*[PowerSync](https://www.powersync.com) is a sync engine for building local-first apps with instantly-responsive UI/UX and simplified state transfer. Syncs between SQLite on the client-side and Postgres or MongoDB on the server-side (MySQL coming soon).*
+_[PowerSync](https://www.powersync.com) is a sync engine for building local-first apps with instantly-responsive UI/UX and simplified state transfer. Syncs between SQLite on the client-side and Postgres or MongoDB on the server-side (MySQL coming soon)._
 
 # PowerSync Service
 

modules/module-mongodb/package.json

@@ -12,7 +12,7 @@
   "scripts": {
     "build": "tsc -b",
     "build:tests": "tsc -b test/tsconfig.json",
-    "clean": "rm -rf ./lib && tsc -b --clean",
+    "clean": "rm -rf ./dist && tsc -b --clean",
     "test": "vitest --no-threads"
   },
   "exports": {
@@ -41,7 +41,7 @@
   "devDependencies": {
     "@types/uuid": "^9.0.4",
     "typescript": "^5.2.2",
-    "vitest": "^0.34.6",
+    "vitest": "^2.1.1",
     "vite-tsconfig-paths": "^4.3.2"
   }
 }

modules/module-mongodb/src/api/MongoRouteAPIAdapter.ts

@@ -33,7 +33,7 @@ export class MongoRouteAPIAdapter implements api.RouteAPI {
     await this.client.close();
   }
 
-  async getSourceConfig(): Promise<service_types.configFile.DataSourceConfig> {
+  async getSourceConfig(): Promise<service_types.configFile.ResolvedDataSourceConfig> {
     return this.config;
   }
 
@@ -165,7 +165,7 @@ export class MongoRouteAPIAdapter implements api.RouteAPI {
     return result;
   }
 
-  async getReplicationLag(syncRulesId: string): Promise<number | undefined> {
+  async getReplicationLag(options: api.ReplicationLagOptions): Promise<number | undefined> {
     // There is no fast way to get replication lag in bytes in MongoDB.
     // We can get replication lag in seconds, but need a different API for that.
     return undefined;

modules/module-mongodb/src/index.ts

@@ -1,5 +1 @@
-import { MongoModule } from './module/MongoModule.js';
-
-export const module = new MongoModule();
-
-export default module;
+export * from './module/MongoModule.js';

modules/module-mongodb/src/module/MongoModule.ts

@@ -4,6 +4,8 @@ import { ConnectionManagerFactory } from '../replication/ConnectionManagerFactor
 import { MongoErrorRateLimiter } from '../replication/MongoErrorRateLimiter.js';
 import { ChangeStreamReplicator } from '../replication/ChangeStreamReplicator.js';
 import * as types from '../types/types.js';
+import { MongoManager } from '../replication/MongoManager.js';
+import { checkSourceConfiguration } from '../replication/replication-utils.js';
 
 export class MongoModule extends replication.ReplicationModule<types.MongoConnectionConfig> {
   constructor() {
@@ -49,4 +51,15 @@ export class MongoModule extends replication.ReplicationModule<types.MongoConnec
   async teardown(options: TearDownOptions): Promise<void> {
     // TODO: Implement?
   }
+
+  async testConnection(config: types.MongoConnectionConfig): Promise<void> {
+    this.decodeConfig(config);
+    const normalisedConfig = this.resolveConfig(this.decodedConfig!);
+    const connectionManager = new MongoManager(normalisedConfig);
+    try {
+      return checkSourceConfiguration(connectionManager);
+    } finally {
+      await connectionManager.end();
+    }
+  }
 }

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

@@ -1,6 +1,6 @@
 import { container, logger } from '@powersync/lib-services-framework';
-import { Metrics, SourceEntityDescriptor, SourceTable, storage } from '@powersync/service-core';
-import { DatabaseInputRow, SqliteRow, SqlSyncRules, TablePattern, toSyncRulesRow } from '@powersync/service-sync-rules';
+import { Metrics, SaveOperationTag, SourceEntityDescriptor, SourceTable, storage } from '@powersync/service-core';
+import { DatabaseInputRow, SqliteRow, SqlSyncRules, TablePattern } from '@powersync/service-sync-rules';
 import * as mongo from 'mongodb';
 import { MongoManager } from './MongoManager.js';
 import {
@@ -248,7 +248,7 @@ export class ChangeStream {
 
       // This auto-flushes when the batch reaches its size limit
       await batch.save({
-        tag: 'insert',
+        tag: SaveOperationTag.INSERT,
         sourceTable: table,
         before: undefined,
         beforeReplicaId: undefined,
@@ -330,7 +330,7 @@ export class ChangeStream {
     if (change.operationType == 'insert') {
       const baseRecord = constructAfterRecord(change.fullDocument);
       return await batch.save({
-        tag: 'insert',
+        tag: SaveOperationTag.INSERT,
         sourceTable: table,
         before: undefined,
         beforeReplicaId: undefined,
@@ -341,15 +341,15 @@ export class ChangeStream {
       if (change.fullDocument == null) {
         // Treat as delete
         return await batch.save({
-          tag: 'delete',
+          tag: SaveOperationTag.DELETE,
           sourceTable: table,
           before: undefined,
           beforeReplicaId: change.documentKey._id
         });
       }
       const after = constructAfterRecord(change.fullDocument!);
       return await batch.save({
-        tag: 'update',
+        tag: SaveOperationTag.UPDATE,
         sourceTable: table,
         before: undefined,
         beforeReplicaId: undefined,
@@ -358,7 +358,7 @@ export class ChangeStream {
       });
     } else if (change.operationType == 'delete') {
       return await batch.save({
-        tag: 'delete',
+        tag: SaveOperationTag.DELETE,
         sourceTable: table,
         before: undefined,
         beforeReplicaId: change.documentKey._id

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

@@ -1,5 +1,4 @@
 import { container } from '@powersync/lib-services-framework';
-import { MongoManager } from './MongoManager.js';
 import { MissingReplicationSlotError, ChangeStream } from './ChangeStream.js';
 
 import { replication } from '@powersync/service-core';
@@ -13,12 +12,10 @@ export interface ChangeStreamReplicationJobOptions extends replication.AbstractR
 
 export class ChangeStreamReplicationJob extends replication.AbstractReplicationJob {
   private connectionFactory: ConnectionManagerFactory;
-  private readonly connectionManager: MongoManager;
 
   constructor(options: ChangeStreamReplicationJobOptions) {
     super(options);
     this.connectionFactory = options.connectionFactory;
-    this.connectionManager = this.connectionFactory.create();
   }
 
   async cleanUp(): Promise<void> {

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

@@ -3,14 +3,14 @@ import { ChangeStreamReplicationJob } from './ChangeStreamReplicationJob.js';
 import { ConnectionManagerFactory } from './ConnectionManagerFactory.js';
 import { MongoErrorRateLimiter } from './MongoErrorRateLimiter.js';
 
-export interface WalStreamReplicatorOptions extends replication.AbstractReplicatorOptions {
+export interface ChangeStreamReplicatorOptions extends replication.AbstractReplicatorOptions {
   connectionFactory: ConnectionManagerFactory;
 }
 
 export class ChangeStreamReplicator extends replication.AbstractReplicator<ChangeStreamReplicationJob> {
   private readonly connectionFactory: ConnectionManagerFactory;
 
-  constructor(options: WalStreamReplicatorOptions) {
+  constructor(options: ChangeStreamReplicatorOptions) {
     super(options);
     this.connectionFactory = options.connectionFactory;
   }

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

@@ -0,0 +1,12 @@
+import * as mongo from 'mongodb';
+import { MongoManager } from './MongoManager.js';
+
+export async function checkSourceConfiguration(connectionManager: MongoManager): Promise<void> {
+  const db = connectionManager.db;
+  const hello = await db.command({ hello: 1 });
+  if (hello.msg == 'isdbgrid') {
+    throw new Error('Sharded MongoDB Clusters are not supported yet (including MongoDB Serverless instances).');
+  } else if (hello.setName == null) {
+    throw new Error('Standalone MongoDB instances are not supported - use a replicaset.');
+  }
+}

modules/module-mongodb/src/types/types.ts

@@ -15,7 +15,7 @@ export interface NormalizedMongoConnectionConfig {
   password?: string;
 }
 
-export const MongoConnectionConfig = service_types.configFile.dataSourceConfig.and(
+export const MongoConnectionConfig = service_types.configFile.DataSourceConfig.and(
   t.object({
     type: t.literal(MONGO_CONNECTION_TYPE),
     /** Unique identifier for the connection - optional when a single connection is present. */

modules/module-mongodb/test/src/change_stream.test.ts

@@ -3,7 +3,7 @@ import { MONGO_STORAGE_FACTORY } from '@core-tests/util.js';
 import { BucketStorageFactory } from '@powersync/service-core';
 import * as crypto from 'crypto';
 import { describe, expect, test } from 'vitest';
-import { walStreamTest } from './change_stream_utils.js';
+import { changeStreamTest } from './change_stream_utils.js';
 import * as mongo from 'mongodb';
 import { setTimeout } from 'node:timers/promises';
 
@@ -19,15 +19,15 @@ bucket_definitions:
 describe(
   'change stream - mongodb',
   function () {
-    defineWalStreamTests(MONGO_STORAGE_FACTORY);
+    defineChangeStreamTests(MONGO_STORAGE_FACTORY);
   },
   { timeout: 20_000 }
 );
 
-function defineWalStreamTests(factory: StorageFactory) {
+function defineChangeStreamTests(factory: StorageFactory) {
   test(
     'replicating basic values',
-    walStreamTest(factory, async (context) => {
+    changeStreamTest(factory, async (context) => {
       const { db } = context;
       await context.updateSyncRules(`
 bucket_definitions:
@@ -66,7 +66,7 @@ bucket_definitions:
 
   test(
     'no fullDocument available',
-    walStreamTest(factory, async (context) => {
+    changeStreamTest(factory, async (context) => {
       const { db, client } = context;
       await context.updateSyncRules(`
 bucket_definitions:
@@ -111,7 +111,7 @@ bucket_definitions:
 
   test(
     'replicating case sensitive table',
-    walStreamTest(factory, async (context) => {
+    changeStreamTest(factory, async (context) => {
       const { db } = context;
       await context.updateSyncRules(`
       bucket_definitions:
@@ -136,7 +136,7 @@ bucket_definitions:
 
   test(
     'replicating large values',
-    walStreamTest(factory, async (context) => {
+    changeStreamTest(factory, async (context) => {
       const { db } = context;
       await context.updateSyncRules(`
       bucket_definitions:
@@ -168,7 +168,7 @@ bucket_definitions:
 
   test(
     'replicating dropCollection',
-    walStreamTest(factory, async (context) => {
+    changeStreamTest(factory, async (context) => {
       const { db } = context;
       const syncRuleContent = `
 bucket_definitions:
@@ -200,7 +200,7 @@ bucket_definitions:
 
   test(
     'replicating renameCollection',
-    walStreamTest(factory, async (context) => {
+    changeStreamTest(factory, async (context) => {
       const { db } = context;
       const syncRuleContent = `
 bucket_definitions:
@@ -232,7 +232,7 @@ bucket_definitions:
 
   test(
     'initial sync',
-    walStreamTest(factory, async (context) => {
+    changeStreamTest(factory, async (context) => {
       const { db } = context;
       await context.updateSyncRules(BASIC_SYNC_RULES);
 
@@ -251,7 +251,7 @@ bucket_definitions:
   // Not correctly implemented yet
   test.skip(
     'large record',
-    walStreamTest(factory, async (context) => {
+    changeStreamTest(factory, async (context) => {
       await context.updateSyncRules(`bucket_definitions:
       global:
         data:
@@ -287,7 +287,7 @@ bucket_definitions:
 
   test(
     'table not in sync rules',
-    walStreamTest(factory, async (context) => {
+    changeStreamTest(factory, async (context) => {
       const { db } = context;
       await context.updateSyncRules(BASIC_SYNC_RULES);
 

modules/module-mongodb/test/src/change_stream_utils.ts

@@ -8,12 +8,12 @@ import * as mongo from 'mongodb';
 import { createCheckpoint } from '@module/replication/MongoRelation.js';
 
 /**
- * Tests operating on the wal stream need to configure the stream and manage asynchronous
+ * Tests operating on the mongo change stream need to configure the stream and manage asynchronous
  * replication, which gets a little tricky.
  *
  * This wraps a test in a function that configures all the context, and tears it down afterwards.
  */
-export function walStreamTest(
+export function changeStreamTest(
   factory: () => Promise<BucketStorageFactory>,
   test: (context: ChangeStreamTestContext) => Promise<void>
 ): () => Promise<void> {

modules/module-mysql/LICENSE

@@ -0,0 +1,67 @@
+# Functional Source License, Version 1.1, Apache 2.0 Future License
+
+## Abbreviation
+
+FSL-1.1-Apache-2.0
+
+## Notice
+
+Copyright 2023-2024 Journey Mobile, Inc.
+
+## Terms and Conditions
+
+### Licensor ("We")
+
+The party offering the Software under these Terms and Conditions.
+
+### The Software
+
+The "Software" is each version of the software that we make available under these Terms and Conditions, as indicated by our inclusion of these Terms and Conditions with the Software.
+
+### License Grant
+
+Subject to your compliance with this License Grant and the Patents, Redistribution and Trademark clauses below, we hereby grant you the right to use, copy, modify, create derivative works, publicly perform, publicly display and redistribute the Software for any Permitted Purpose identified below.
+
+### Permitted Purpose
+
+A Permitted Purpose is any purpose other than a Competing Use. A Competing Use means making the Software available to others in a commercial product or service that:
+
+1. substitutes for the Software;
+2. substitutes for any other product or service we offer using the Software that exists as of the date we make the Software available; or
+3. offers the same or substantially similar functionality as the Software.
+
+Permitted Purposes specifically include using the Software:
+
+1. for your internal use and access;
+2. for non-commercial education;
+3. for non-commercial research; and
+4. in connection with professional services that you provide to a licensee using the Software in accordance with these Terms and Conditions.
+
+### Patents
+
+To the extent your use for a Permitted Purpose would necessarily infringe our patents, the license grant above includes a license under our patents. If you make a claim against any party that the Software infringes or contributes to the infringement of any patent, then your patent license to the Software ends immediately.
+
+### Redistribution
+
+The Terms and Conditions apply to all copies, modifications and derivatives of the Software.
+If you redistribute any copies, modifications or derivatives of the Software, you must include a copy of or a link to these Terms and Conditions and not remove any copyright notices provided in or with the Software.
+
+### Disclaimer
+
+THE SOFTWARE IS PROVIDED "AS IS" AND WITHOUT WARRANTIES OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION WARRANTIES OF FITNESS FOR A PARTICULAR PURPOSE, MERCHANTABILITY, TITLE OR NON-INFRINGEMENT.
+IN NO EVENT WILL WE HAVE ANY LIABILITY TO YOU ARISING OUT OF OR RELATED TO THE SOFTWARE, INCLUDING INDIRECT, SPECIAL, INCIDENTAL OR CONSEQUENTIAL DAMAGES, EVEN IF WE HAVE BEEN INFORMED OF THEIR POSSIBILITY IN ADVANCE.
+
+### Trademarks
+
+Except for displaying the License Details and identifying us as the origin of the Software, you have no right under these Terms and Conditions to use our trademarks, trade names, service marks or product names.
+
+## Grant of Future License
+
+We hereby irrevocably grant you an additional license to use the Software under the Apache License, Version 2.0 that is effective on the second anniversary of the date we make the Software available. On or after that date, you may use the Software under the Apache License, Version 2.0, in which case the following will apply:
+
+Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

modules/module-mysql/README.md

@@ -0,0 +1,3 @@
+# PowerSync MySQL Module
+
+This is a module which provides MySQL replication to PowerSync.

modules/module-mysql/dev/.env.template

@@ -0,0 +1,2 @@
+PS_MONGO_URI=mongodb://mongo:27017/powersync_demo
+PS_PORT=8080