Merge branch 'raw_tables_skilldevs' into my-example

Author: davidmartos96

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

@@ -44,6 +44,9 @@ class PowerSyncDatabaseImpl
   @override
   SqliteDatabase database;
 
+  @override
+  bool manualSchemaManagement;
+
   @override
   @protected
   late Future<void> isInitialized;
@@ -76,15 +79,21 @@ class PowerSyncDatabaseImpl
       required String path,
       int maxReaders = SqliteDatabase.defaultMaxReaders,
       Logger? logger,
+      bool manualSchemaManagement = false,
       @Deprecated("Use [PowerSyncDatabase.withFactory] instead.")
       // ignore: deprecated_member_use_from_same_package
       SqliteConnectionSetup? sqliteSetup}) {
     // ignore: deprecated_member_use_from_same_package
     DefaultSqliteOpenFactory factory =
         // ignore: deprecated_member_use_from_same_package
         PowerSyncOpenFactory(path: path, sqliteSetup: sqliteSetup);
-    return PowerSyncDatabaseImpl.withFactory(factory,
-        schema: schema, maxReaders: maxReaders, logger: logger);
+    return PowerSyncDatabaseImpl.withFactory(
+      factory,
+      schema: schema,
+      maxReaders: maxReaders,
+      logger: logger,
+      manualSchemaManagement: manualSchemaManagement,
+    );
   }
 
   /// Open a [PowerSyncDatabase] with a [PowerSyncOpenFactory].
@@ -96,22 +105,32 @@ class PowerSyncDatabaseImpl
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.
   factory PowerSyncDatabaseImpl.withFactory(
-      DefaultSqliteOpenFactory openFactory,
-      {required Schema schema,
-      int maxReaders = SqliteDatabase.defaultMaxReaders,
-      Logger? logger}) {
+    DefaultSqliteOpenFactory openFactory, {
+    required Schema schema,
+    int maxReaders = SqliteDatabase.defaultMaxReaders,
+    Logger? logger,
+    bool manualSchemaManagement = false,
+  }) {
     final db = SqliteDatabase.withFactory(openFactory, maxReaders: maxReaders);
     return PowerSyncDatabaseImpl.withDatabase(
-        schema: schema, database: db, logger: logger);
+      schema: schema,
+      database: db,
+      logger: logger,
+      manualSchemaManagement: manualSchemaManagement,
+    );
   }
 
   /// Open a PowerSyncDatabase on an existing [SqliteDatabase].
   ///
   /// Migrations are run on the database when this constructor is called.
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.s
-  PowerSyncDatabaseImpl.withDatabase(
-      {required this.schema, required this.database, Logger? logger}) {
+  PowerSyncDatabaseImpl.withDatabase({
+    required this.schema,
+    required this.database,
+    Logger? logger,
+    this.manualSchemaManagement = false,
+  }) {
     this.logger = logger ?? autoLogger;
     isInitialized = baseInit();
   }
@@ -120,7 +139,7 @@ class PowerSyncDatabaseImpl
   @internal
   Future<void> connectInternal({
     required PowerSyncBackendConnector connector,
-    required SyncOptions options,
+    required ResolvedSyncOptions options,
     required AbortController abort,
     required Zone asyncWorkZone,
   }) async {
@@ -135,7 +154,6 @@ class PowerSyncDatabaseImpl
     SendPort? initPort;
     final hasInitPort = Completer<void>();
     final receivedIsolateExit = Completer<void>();
-    final resolved = ResolvedSyncOptions(options);
 
     Future<void> waitForShutdown() async {
       // Only complete the abortion signal after the isolate shuts down. This
@@ -183,7 +201,7 @@ class PowerSyncDatabaseImpl
           final port = initPort = data[1] as SendPort;
           hasInitPort.complete();
           var crudStream = database
-              .onChange(['ps_crud'], throttle: resolved.crudThrottleTime);
+              .onChange(['ps_crud'], throttle: options.crudThrottleTime);
           crudUpdateSubscription = crudStream.listen((event) {
             port.send(['update']);
           });
@@ -245,9 +263,10 @@ class PowerSyncDatabaseImpl
       _PowerSyncDatabaseIsolateArgs(
         receiveMessages.sendPort,
         dbRef,
-        resolved,
+        options,
         crudMutex.shared,
         syncMutex.shared,
+        schema,
       ),
       debugName: 'Sync ${database.openFactory.path}',
       onError: receiveUnhandledErrors.sendPort,
@@ -291,13 +310,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,
   );
 }
 
@@ -393,6 +414,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/powersync_database.dart

@@ -32,19 +32,23 @@ abstract class PowerSyncDatabase
   /// A maximum of [maxReaders] concurrent read transactions are allowed.
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.
-  factory PowerSyncDatabase(
-      {required Schema schema,
-      required String path,
-      Logger? logger,
-      @Deprecated("Use [PowerSyncDatabase.withFactory] instead.")
-      // ignore: deprecated_member_use_from_same_package
-      SqliteConnectionSetup? sqliteSetup}) {
+  factory PowerSyncDatabase({
+    required Schema schema,
+    required String path,
+    Logger? logger,
+    bool manualSchemaManagement = false,
+    @Deprecated("Use [PowerSyncDatabase.withFactory] instead.")
+    // ignore: deprecated_member_use_from_same_package
+    SqliteConnectionSetup? sqliteSetup,
+  }) {
     return PowerSyncDatabaseImpl(
-        schema: schema,
-        path: path,
-        logger: logger,
-        // ignore: deprecated_member_use_from_same_package
-        sqliteSetup: sqliteSetup);
+      schema: schema,
+      path: path,
+      manualSchemaManagement: manualSchemaManagement,
+      logger: logger,
+      // ignore: deprecated_member_use_from_same_package
+      sqliteSetup: sqliteSetup,
+    );
   }
 
   /// Open a [PowerSyncDatabase] with a [PowerSyncOpenFactory].
@@ -55,24 +59,38 @@ abstract class PowerSyncDatabase
   /// Subclass [PowerSyncOpenFactory] to add custom logic to this process.
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.
-  factory PowerSyncDatabase.withFactory(DefaultSqliteOpenFactory openFactory,
-      {required Schema schema,
-      int maxReaders = SqliteDatabase.defaultMaxReaders,
-      Logger? logger}) {
-    return PowerSyncDatabaseImpl.withFactory(openFactory,
-        schema: schema, maxReaders: maxReaders, logger: logger);
+  factory PowerSyncDatabase.withFactory(
+    DefaultSqliteOpenFactory openFactory, {
+    required Schema schema,
+    int maxReaders = SqliteDatabase.defaultMaxReaders,
+    bool manualSchemaManagement = false,
+    Logger? logger,
+  }) {
+    return PowerSyncDatabaseImpl.withFactory(
+      openFactory,
+      schema: schema,
+      maxReaders: maxReaders,
+      manualSchemaManagement: manualSchemaManagement,
+      logger: logger,
+    );
   }
 
   /// Open a PowerSyncDatabase on an existing [SqliteDatabase].
   ///
   /// Migrations are run on the database when this constructor is called.
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.
-  factory PowerSyncDatabase.withDatabase(
-      {required Schema schema,
-      required SqliteDatabase database,
-      Logger? loggers}) {
+  factory PowerSyncDatabase.withDatabase({
+    required Schema schema,
+    required SqliteDatabase database,
+    bool manualSchemaManagement = false,
+    Logger? logger,
+  }) {
     return PowerSyncDatabaseImpl.withDatabase(
-        schema: schema, database: database, logger: loggers);
+      schema: schema,
+      database: database,
+      manualSchemaManagement: manualSchemaManagement,
+      logger: logger,
+    );
   }
 }

packages/powersync_core/lib/src/database/powersync_database_impl_stub.dart

@@ -32,6 +32,9 @@ class PowerSyncDatabaseImpl
   @override
   SqliteDatabase get database => throw UnimplementedError();
 
+  @override
+  bool get manualSchemaManagement => throw UnimplementedError();
+
   @override
   Future<void> get isInitialized => throw UnimplementedError();
 
@@ -53,6 +56,7 @@ class PowerSyncDatabaseImpl
       {required Schema schema,
       required String path,
       int maxReaders = SqliteDatabase.defaultMaxReaders,
+      bool manualSchemaManagement = false,
       Logger? logger,
       @Deprecated("Use [PowerSyncDatabase.withFactory] instead.")
       // ignore: deprecated_member_use_from_same_package
@@ -72,6 +76,7 @@ class PowerSyncDatabaseImpl
     DefaultSqliteOpenFactory openFactory, {
     required Schema schema,
     int maxReaders = SqliteDatabase.defaultMaxReaders,
+    bool manualSchemaManagement = false,
     Logger? logger,
   }) {
     throw UnimplementedError();
@@ -82,10 +87,12 @@ class PowerSyncDatabaseImpl
   /// Migrations are run on the database when this constructor is called.
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.s
-  factory PowerSyncDatabaseImpl.withDatabase(
-      {required Schema schema,
-      required SqliteDatabase database,
-      Logger? logger}) {
+  factory PowerSyncDatabaseImpl.withDatabase({
+    required Schema schema,
+    required SqliteDatabase database,
+    bool manualSchemaManagement = false,
+    Logger? logger,
+  }) {
     throw UnimplementedError();
   }
 
@@ -115,7 +122,7 @@ class PowerSyncDatabaseImpl
     required PowerSyncBackendConnector connector,
     required AbortController abort,
     required Zone asyncWorkZone,
-    required SyncOptions options,
+    required ResolvedSyncOptions options,
   }) {
     throw UnimplementedError();
   }

packages/powersync_core/lib/src/database/powersync_db_mixin.dart

@@ -38,6 +38,10 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// Use [attachedLogger] to propagate logs to [Logger.root] for custom logging.
   Logger get logger;
 
+  bool get manualSchemaManagement;
+
+  bool _manualSchemaManagementCompleted = false;
+
   @Deprecated("This field is unused, pass params to connect() instead")
   Map<String, dynamic>? clientParams;
 
@@ -110,10 +114,36 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     statusStream = statusStreamController.stream;
     updates = powerSyncUpdateNotifications(database.updates);
 
+    _manualSchemaManagementCompleted = false;
+
     await database.initialize();
     await _checkVersion();
     await database.execute('SELECT powersync_init()');
-    await updateSchema(schema);
+
+    if (!manualSchemaManagement) {
+      // Create the internal db schema
+      await updateSchema(schema);
+      await _afterSchemaReady();
+    }
+  }
+
+  Future<void> markSchemaAsReady() async {
+    await isInitialized;
+    _manualSchemaManagementCompleted = true;
+
+    await _afterSchemaReady();
+  }
+
+  void _assertSchemaIsReady() {
+    if (!manualSchemaManagement || _manualSchemaManagementCompleted) {
+      return;
+    }
+
+    throw AssertionError(
+        'In manual schema management mode, you need to mark the powersync database as ready');
+  }
+
+  Future<void> _afterSchemaReady() async {
     await _updateHasSynced();
   }
 
@@ -289,11 +319,14 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     // the lock for the connection.
     await initialize();
 
-    final resolvedOptions = SyncOptions(
-      crudThrottleTime: options?.crudThrottleTime ?? crudThrottleTime,
+    _assertSchemaIsReady();
+
+    final resolvedOptions = ResolvedSyncOptions.resolve(
+      options,
+      crudThrottleTime: crudThrottleTime,
       // ignore: deprecated_member_use_from_same_package
-      retryDelay: options?.retryDelay ?? retryDelay,
-      params: options?.params ?? params,
+      retryDelay: retryDelay,
+      params: params,
     );
 
     // ignore: deprecated_member_use_from_same_package
@@ -362,7 +395,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   @internal
   Future<void> connectInternal({
     required PowerSyncBackendConnector connector,
-    required SyncOptions options,
+    required ResolvedSyncOptions options,
     required AbortController abort,
     required Zone asyncWorkZone,
   });
@@ -451,13 +484,15 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// Get an unique id for this client.
   /// This id is only reset when the database is deleted.
   Future<String> getClientId() async {
+    _assertSchemaIsReady(); // TODO(skilldevs): Needed?
     final row = await get('SELECT powersync_client_id() as client_id');
     return row['client_id'] as String;
   }
 
   /// Get upload queue size estimate and count.
   Future<UploadQueueStats> getUploadQueueStats(
       {bool includeSize = false}) async {
+    _assertSchemaIsReady();
     if (includeSize) {
       final row = await getOptional(
           'SELECT SUM(cast(data as blob) + 20) as size, count(*) as count FROM ps_crud');
@@ -485,6 +520,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// data by transaction. One batch may contain data from multiple transactions,
   /// and a single transaction may be split over multiple batches.
   Future<CrudBatch?> getCrudBatch({int limit = 100}) async {
+    _assertSchemaIsReady();
     final rows = await getAll(
         'SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT ?',
         [limit + 1]);
@@ -531,6 +567,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// Unlike [getCrudBatch], this only returns data from a single transaction at a time.
   /// All data for the transaction is loaded into memory.
   Future<CrudTransaction?> getNextCrudTransaction() async {
+    _assertSchemaIsReady();
     return await readTransaction((tx) async {
       final first = await tx.getOptional(
           'SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT 1');