pass schema to sync worker

Author: js2702

packages/powersync_core/lib/src/database/native/native_powersync_database.dart

@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:convert';
 import 'dart:isolate';
 import 'package:meta/meta.dart';
 
@@ -266,7 +267,7 @@ class PowerSyncDatabaseImpl
         options,
         crudMutex.shared,
         syncMutex.shared,
-        schema,
+        jsonEncode(schema),
       ),
       debugName: 'Sync ${database.openFactory.path}',
       onError: receiveUnhandledErrors.sendPort,
@@ -310,15 +311,15 @@ class _PowerSyncDatabaseIsolateArgs {
   final ResolvedSyncOptions options;
   final SerializedMutex crudMutex;
   final SerializedMutex syncMutex;
-  final Schema schema;
+  final String schemaJson;
 
   _PowerSyncDatabaseIsolateArgs(
     this.sPort,
     this.dbRef,
     this.options,
     this.crudMutex,
     this.syncMutex,
-    this.schema,
+    this.schemaJson,
   );
 }
 
@@ -414,7 +415,7 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
     final storage = BucketStorage(connection);
     final sync = StreamingSyncImplementation(
       adapter: storage,
-      schema: args.schema,
+      schemaJson: args.schemaJson,
       connector: InternalConnector(
         getCredentialsCached: getCredentialsCached,
         prefetchCredentials: prefetchCredentials,

packages/powersync_core/lib/src/database/web/web_powersync_database.dart

@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:convert';
 import 'package:meta/meta.dart';
 import 'package:http/browser_client.dart';
 import 'package:logging/logging.dart';
@@ -159,7 +160,7 @@ class PowerSyncDatabaseImpl
 
       sync = StreamingSyncImplementation(
         adapter: storage,
-        schema: schema,
+        schemaJson: jsonEncode(schema),
         connector: InternalConnector.wrap(connector, this),
         crudUpdateTriggerStream: crudStream,
         options: options,

packages/powersync_core/lib/src/sync/streaming_sync.dart

@@ -5,7 +5,6 @@ import 'dart:typed_data';
 import 'package:http/http.dart' as http;
 import 'package:logging/logging.dart';
 import 'package:meta/meta.dart';
-import 'package:powersync_core/powersync_core.dart';
 import 'package:powersync_core/src/abort_controller.dart';
 import 'package:powersync_core/src/exceptions.dart';
 import 'package:powersync_core/src/log_internal.dart';
@@ -33,7 +32,7 @@ abstract interface class StreamingSync {
 
 @internal
 class StreamingSyncImplementation implements StreamingSync {
-  final Schema? schema; //TODO(SkillDevs): pass in all implementations
+  final String schemaJson;
   final BucketStorage adapter;
   final InternalConnector connector;
   final ResolvedSyncOptions options;
@@ -64,7 +63,7 @@ class StreamingSyncImplementation implements StreamingSync {
   String? clientId;
 
   StreamingSyncImplementation({
-    required this.schema,
+    required this.schemaJson,
     required this.adapter,
     required this.connector,
     required this.crudUpdateTriggerStream,
@@ -602,7 +601,7 @@ final class _ActiveRustStreamingIteration {
         'start',
         convert.json.encode({
           'parameters': sync.options.params,
-          'schema': sync.schema,
+          'schema': convert.json.decode(sync.schemaJson),
         }),
       );
       assert(_completedStream.isCompleted, 'Should have started streaming');

packages/powersync_core/lib/src/web/sync_controller.dart

@@ -113,6 +113,9 @@ class SyncWorkerHandle implements StreamingSync {
   @override
   Future<void> streamingSync() async {
     await _channel.startSynchronization(
-        database.database.openFactory.path, ResolvedSyncOptions(options));
+      database.database.openFactory.path,
+      ResolvedSyncOptions(options),
+      database.schema,
+    );
   }
 }

packages/powersync_core/lib/src/web/sync_worker.dart

@@ -45,12 +45,16 @@ class _SyncWorker {
     });
   }
 
-  _SyncRunner referenceSyncTask(
-      String databaseIdentifier, SyncOptions options, _ConnectedClient client) {
+  _SyncRunner referenceSyncTask(String databaseIdentifier, SyncOptions options,
+      String schemaJson, _ConnectedClient client) {
     return _requestedSyncTasks.putIfAbsent(databaseIdentifier, () {
       return _SyncRunner(databaseIdentifier);
     })
-      ..registerClient(client, options);
+      ..registerClient(
+        client,
+        options,
+        schemaJson,
+      );
   }
 }
 
@@ -86,8 +90,8 @@ class _ConnectedClient {
               },
             );
 
-            _runner = _worker.referenceSyncTask(
-                request.databaseName, recoveredOptions, this);
+            _runner = _worker.referenceSyncTask(request.databaseName,
+                recoveredOptions, request.schemaJson, this);
             return (JSObject(), null);
           case SyncWorkerMessageType.abortSynchronization:
             _runner?.disconnectClient(this);
@@ -128,6 +132,7 @@ class _ConnectedClient {
 class _SyncRunner {
   final String identifier;
   ResolvedSyncOptions options = ResolvedSyncOptions(SyncOptions());
+  String schemaJson = '{}';
 
   final StreamGroup<_RunnerEvent> _group = StreamGroup();
   final StreamController<_RunnerEvent> _mainEvents = StreamController();
@@ -146,10 +151,12 @@ class _SyncRunner {
             case _AddConnection(
                 :final client,
                 :final options,
+                :final schemaJson,
               ):
               connections.add(client);
               final (newOptions, reconnect) = this.options.applyFrom(options);
               this.options = newOptions;
+              this.schemaJson = schemaJson;
 
               if (sync == null) {
                 await _requestDatabase(client);
@@ -264,7 +271,7 @@ class _SyncRunner {
 
     sync = StreamingSyncImplementation(
       adapter: WebBucketStorage(database),
-      schema: null,
+      schemaJson: client._runner!.schemaJson,
       connector: InternalConnector(
         getCredentialsCached: client.channel.credentialsCallback,
         prefetchCredentials: ({required bool invalidate}) async {
@@ -287,8 +294,9 @@ class _SyncRunner {
     sync!.streamingSync();
   }
 
-  void registerClient(_ConnectedClient client, SyncOptions options) {
-    _mainEvents.add(_AddConnection(client, options));
+  void registerClient(
+      _ConnectedClient client, SyncOptions options, String schemaJson) {
+    _mainEvents.add(_AddConnection(client, options, schemaJson));
   }
 
   /// Remove a client, disconnecting if no clients remain..
@@ -307,8 +315,9 @@ sealed class _RunnerEvent {}
 final class _AddConnection implements _RunnerEvent {
   final _ConnectedClient client;
   final SyncOptions options;
+  final String schemaJson;
 
-  _AddConnection(this.client, this.options);
+  _AddConnection(this.client, this.options, this.schemaJson);
 }
 
 final class _RemoveConnection implements _RunnerEvent {

packages/powersync_core/lib/src/web/sync_worker_protocol.dart

@@ -3,6 +3,7 @@ import 'dart:convert';
 import 'dart:js_interop';
 
 import 'package:logging/logging.dart';
+import 'package:powersync_core/src/schema.dart';
 import 'package:powersync_core/src/sync/options.dart';
 import 'package:web/web.dart';
 
@@ -71,6 +72,7 @@ extension type StartSynchronization._(JSObject _) implements JSObject {
     required int requestId,
     required int retryDelayMs,
     required String implementationName,
+    required String schemaJson,
     String? syncParamsEncoded,
   });
 
@@ -79,6 +81,7 @@ extension type StartSynchronization._(JSObject _) implements JSObject {
   external int get crudThrottleTimeMs;
   external int? get retryDelayMs;
   external String? get implementationName;
+  external String get schemaJson;
   external String? get syncParamsEncoded;
 }
 
@@ -410,7 +413,7 @@ final class WorkerCommunicationChannel {
   }
 
   Future<void> startSynchronization(
-      String databaseName, ResolvedSyncOptions options) async {
+      String databaseName, ResolvedSyncOptions options, Schema schema) async {
     final (id, completion) = _newRequest();
     port.postMessage(SyncWorkerMessage(
       type: SyncWorkerMessageType.startSynchronization.name,
@@ -420,6 +423,7 @@ final class WorkerCommunicationChannel {
         retryDelayMs: options.retryDelay.inMilliseconds,
         requestId: id,
         implementationName: options.source.syncImplementation.name,
+        schemaJson: jsonEncode(schema),
         syncParamsEncoded: switch (options.source.params) {
           null => null,
           final params => jsonEncode(params),

packages/powersync_core/test/utils/abstract_test_utils.dart

@@ -1,3 +1,5 @@
+import 'dart:convert';
+
 import 'package:http/http.dart';
 import 'package:logging/logging.dart';
 import 'package:powersync_core/powersync_core.dart';
@@ -154,7 +156,7 @@ extension MockSync on PowerSyncDatabase {
   }) {
     final impl = StreamingSyncImplementation(
       adapter: BucketStorage(this),
-      schema: schema,
+      schemaJson: jsonEncode(schema),
       client: client,
       options: ResolvedSyncOptions(options),
       connector: InternalConnector.wrap(connector, this),