raw tables in schema

Author: js2702

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

@@ -247,6 +247,7 @@ class PowerSyncDatabaseImpl
         options,
         crudMutex.shared,
         syncMutex.shared,
+        schema,
       ),
       debugName: 'Sync ${database.openFactory.path}',
       onError: receiveUnhandledErrors.sendPort,
@@ -290,13 +291,15 @@ class _PowerSyncDatabaseIsolateArgs {
   final ResolvedSyncOptions options;
   final SerializedMutex crudMutex;
   final SerializedMutex syncMutex;
+  final Schema schema;
 
   _PowerSyncDatabaseIsolateArgs(
     this.sPort,
     this.dbRef,
     this.options,
     this.crudMutex,
     this.syncMutex,
+    this.schema,
   );
 }
 
@@ -392,6 +395,7 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
     final storage = BucketStorage(connection);
     final sync = StreamingSyncImplementation(
       adapter: storage,
+      schema: args.schema,
       connector: InternalConnector(
         getCredentialsCached: getCredentialsCached,
         prefetchCredentials: prefetchCredentials,

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

@@ -141,6 +141,7 @@ class PowerSyncDatabaseImpl
 
       sync = StreamingSyncImplementation(
         adapter: storage,
+        schema: schema,
         connector: InternalConnector.wrap(connector, this),
         crudUpdateTriggerStream: crudStream,
         options: options,

packages/powersync_core/lib/src/schema.dart

@@ -8,10 +8,11 @@ import 'schema_logic.dart';
 class Schema {
   /// List of tables in the schema.
   final List<Table> tables;
+  final List<RawTable> rawTables;
 
-  const Schema(this.tables);
+  const Schema(this.tables, {this.rawTables = const []});
 
-  Map<String, dynamic> toJson() => {'tables': tables};
+  Map<String, dynamic> toJson() => {'raw_tables': rawTables, 'tables': tables};
 
   void validate() {
     Set<String> tableNames = {};
@@ -315,6 +316,60 @@ class Column {
   Map<String, dynamic> toJson() => {'name': name, 'type': type.sqlite};
 }
 
+class RawTable {
+  final String
+      name; // TODO: it does not need to be the same name as the raw table
+  final PendingStatement put;
+  final PendingStatement delete;
+
+  const RawTable(
+    this.name,
+    this.put,
+    this.delete,
+  );
+
+  Map<String, dynamic> toJson() => {
+        'name': name,
+        'put': put,
+        'delete': delete,
+      };
+}
+
+class PendingStatement {
+  final String sql;
+  final List<PendingStatementValue> params;
+
+  PendingStatement({required this.sql, required this.params});
+
+  Map<String, dynamic> toJson() => {
+        'sql': sql,
+        'params': params,
+      };
+}
+
+sealed class PendingStatementValue {
+  dynamic toJson();
+}
+
+class PendingStmtValueColumn extends PendingStatementValue {
+  final String column;
+  PendingStmtValueColumn(this.column);
+
+  @override
+  dynamic toJson() {
+    return {
+      'Column': column,
+    };
+  }
+}
+
+class PendingStmtValueId extends PendingStatementValue {
+  @override
+  dynamic toJson() {
+    return 'Id';
+  }
+}
+
 /// Type of column.
 enum ColumnType {
   /// TEXT column.

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

@@ -5,6 +5,7 @@ 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';
@@ -32,6 +33,7 @@ abstract interface class StreamingSync {
 
 @internal
 class StreamingSyncImplementation implements StreamingSync {
+  final Schema? schema; //TODO(SkillDevs): pass in all implementations
   final BucketStorage adapter;
   final InternalConnector connector;
   final ResolvedSyncOptions options;
@@ -62,6 +64,7 @@ class StreamingSyncImplementation implements StreamingSync {
   String? clientId;
 
   StreamingSyncImplementation({
+    required this.schema,
     required this.adapter,
     required this.connector,
     required this.crudUpdateTriggerStream,
@@ -592,7 +595,7 @@ final class _ActiveRustStreamingIteration {
         'start',
         convert.json.encode({
           'parameters': sync.options.params,
-          'schema': 'TODO: Pass-through schema (probably in serialized form)',
+          'schema': sync.schema,
         }),
       );
       assert(_completedStream.isCompleted, 'Should have started streaming');

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

@@ -264,6 +264,7 @@ class _SyncRunner {
 
     sync = StreamingSyncImplementation(
       adapter: WebBucketStorage(database),
+      schema: null,
       connector: InternalConnector(
         getCredentialsCached: client.channel.credentialsCallback,
         prefetchCredentials: ({required bool invalidate}) async {

packages/powersync_core/test/utils/abstract_test_utils.dart

@@ -153,6 +153,7 @@ extension MockSync on PowerSyncDatabase {
   }) {
     final impl = StreamingSyncImplementation(
       adapter: BucketStorage(this),
+      schema: null,
       client: client,
       options: ResolvedSyncOptions(options),
       connector: InternalConnector.wrap(connector, this),