Merge pull request #102 from SkillDevs/custom_stream_handling

Allow transforming table updates from sqlite_async.

Author: simolus3

packages/drift_sqlite_async/CHANGELOG.md

@@ -1,3 +1,7 @@
+## 0.2.4
+
+- Allow transforming table updates from sqlite_async.
+
 ## 0.2.3+1
 
  - Update a dependency to the latest release.

packages/drift_sqlite_async/lib/src/connection.dart

@@ -15,12 +15,22 @@ import 'package:sqlite_async/sqlite_async.dart';
 class SqliteAsyncDriftConnection extends DatabaseConnection {
   late StreamSubscription _updateSubscription;
 
-  SqliteAsyncDriftConnection(SqliteConnection db, {bool logStatements = false})
-      : super(SqliteAsyncQueryExecutor(db, logStatements: logStatements)) {
+  /// [transformTableUpdates] is useful to map local table names from PowerSync that are backed by a view name
+  /// which is the entity that the user interacts with.
+  SqliteAsyncDriftConnection(
+    SqliteConnection db, {
+    bool logStatements = false,
+    Set<TableUpdate> Function(UpdateNotification)? transformTableUpdates,
+  }) : super(SqliteAsyncQueryExecutor(db, logStatements: logStatements)) {
     _updateSubscription = (db as SqliteQueries).updates!.listen((event) {
-      var setUpdates = <TableUpdate>{};
-      for (var tableName in event.tables) {
-        setUpdates.add(TableUpdate(tableName));
+      final Set<TableUpdate> setUpdates;
+      if (transformTableUpdates != null) {
+        setUpdates = transformTableUpdates(event);
+      } else {
+        setUpdates = <TableUpdate>{};
+        for (var tableName in event.tables) {
+          setUpdates.add(TableUpdate(tableName));
+        }
       }
       super.streamQueries.handleTableUpdates(setUpdates);
     });

packages/drift_sqlite_async/pubspec.yaml

@@ -1,5 +1,5 @@
 name: drift_sqlite_async
-version: 0.2.3+1
+version: 0.2.4
 homepage: https://github.com/powersync-ja/sqlite_async.dart
 repository: https://github.com/powersync-ja/sqlite_async.dart
 description: Use Drift with a sqlite_async database, allowing both to be used in the same application.

packages/drift_sqlite_async/test/basic_test.dart

@@ -11,6 +11,7 @@ import 'package:sqlite_async/sqlite_async.dart';
 import 'package:test/test.dart';
 
 import './utils/test_utils.dart';
+import 'generated/database.dart';
 
 class EmptyDatabase extends GeneratedDatabase {
   EmptyDatabase(super.executor);
@@ -245,4 +246,43 @@ INSERT INTO test_data(description) VALUES('test data');
       expect(row, isEmpty);
     });
   });
+
+  test('transform table updates', () async {
+    final path = dbPath();
+    await cleanDb(path: path);
+
+    final db = await setupDatabase(path: path);
+    final connection = SqliteAsyncDriftConnection(
+      db,
+      // tables with the local_ prefix are mapped to the name without the prefix
+      transformTableUpdates: (event) {
+        final updates = <TableUpdate>{};
+
+        for (final originalTableName in event.tables) {
+          final effectiveName = originalTableName.startsWith("local_")
+              ? originalTableName.substring(6)
+              : originalTableName;
+          updates.add(TableUpdate(effectiveName));
+        }
+
+        return updates;
+      },
+    );
+
+    // Create table with a different name than drift. (Mimicking a table name backed by a view in PowerSync with the optional sync strategy)
+    await db.execute(
+      'CREATE TABLE local_todos(id INTEGER PRIMARY KEY AUTOINCREMENT, description TEXT)',
+    );
+
+    final dbu = TodoDatabase.fromSqliteAsyncConnection(connection);
+
+    final tableUpdatesFut =
+        dbu.tableUpdates(TableUpdateQuery.onTableName("todos")).first;
+
+    // This insert will trigger the sqlite_async "updates" stream
+    await db.execute("INSERT INTO local_todos(description) VALUES('Test 1')");
+
+    expect(await tableUpdatesFut.timeout(const Duration(seconds: 2)),
+        {TableUpdate("todos")});
+  });
 }

packages/drift_sqlite_async/test/generated/database.dart

@@ -15,6 +15,8 @@ class TodoItems extends Table {
 @DriftDatabase(tables: [TodoItems])
 class TodoDatabase extends _$TodoDatabase {
   TodoDatabase(SqliteConnection db) : super(SqliteAsyncDriftConnection(db));
+  
+  TodoDatabase.fromSqliteAsyncConnection(SqliteAsyncDriftConnection super.conn);
 
   @override
   int get schemaVersion => 1;