fix: Close generated sync sessions and expose the shared demo (#135)

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## Unreleased
 
 - fix: Keep non-synced `updateRow` calls in the test harness transaction.
+- fix: Allow generated client sync sessions to close their SQLite connections.
 - fix: Resolve shared-package models and enums during projection and sync.
 - fix: Preserve boolean types when syncing SQLite column updates.
 - fix: Preserve JSON and JSONB field types and values through inserts, updates,

README.md

@@ -175,6 +175,14 @@ await client.offlineSync.syncOnce(session);
 final syncSession = client.offlineSync.syncContinuously(session);
 ```
 
+When disposing a client replica, cancel its sync subscription before closing
+the local database:
+
+```dart
+await syncSession.cancel();
+await session.close();
+```
+
 ## Usage
 
 Once wired up, use normal generated-model CRUD against the `session` instance,

example/offline_sync_demo/README.md

@@ -9,23 +9,25 @@ them converge.
 
 If you are here to learn the package, read **`lib/src/offline_replica.dart`**
 first — it is the entire integration in ~60 lines, deliberately free of demo and
-UI code. Every replica follows the same four steps:
+UI code. Every replica follows the same three steps:
 
-1. **Open a local database.** `client.createSession(path)` opens a Serverpod
+1. **Open a sync session.** The generated
-   client database (a local SQLite file) and runs the client migrations.
+   `client.createSyncSession(path, isDebugMode: kDebugMode, persistentUserId: …)`
-2. **Wrap it for CRDT sync.** `OfflineSyncDatabaseSession.wraps(raw, syncTables: …,
+   opens the local SQLite database, runs client migrations, wraps the generated
-   persistentUserId: …)`, then `offlineSyncSession.db.initialize()` to establish this
+   sync tables, and initializes CRDT tracking.
-   device's CRDT node so it can take part in sync.
+2. **Read and write generated models** against that session, e.g.
-3. **Read and write generated models** against that session, e.g.
    `Person.db.insertRow(session, person)` or `session.db.find<Person>()`. (The
    demo also routes these through a generic `TableOps` registry so its
    metadata-driven UI can touch any table without a per-type switch — that part
    is demo-only; the `seed*` methods in `lib/src/demo_controller.dart` show the
    plain form.)
-4. **Synchronize through a Serverpod client.**
+3. **Synchronize through a Serverpod client.**
    `client.offlineSync.syncOnce(session, onMergeSuccess: …)` for a one-shot push/pull,
    or `client.offlineSync.syncContinuously(session, …)` to stream until cancelled.
 
+When disposing a replica, cancel its sync subscription and call `session.close()`
+to release the local database connection.
+
 Connectivity is just *which client you sync through*: the demo swaps
 `Client.httpClientOverride` between a real transport and a failing one to
 simulate going offline — the local database is untouched either way.

example/offline_sync_demo/lib/src/demo_controller.dart

@@ -191,10 +191,7 @@ class DemoController extends ChangeNotifier {
   void dispose() {
     _disposed = true;
     stopScenarioAutoPlay();
-    unawaited(_stopAllStreams());
+    unawaited(_closeSessions());
-    for (final replica in _sessions.values) {
-      unawaited(replica.close());
-    }
     for (final state in replicas.values) {
       state.dispose();
     }
@@ -202,6 +199,14 @@ class DemoController extends ChangeNotifier {
     super.dispose();
   }
 
+  Future<void> _closeSessions() async {
+    try {
+      await _stopAllStreams();
+    } finally {
+      await Future.wait(_sessions.values.map((replica) => replica.close()));
+    }
+  }
+
   @override
   void notifyListeners() {
     if (_disposed) return;

example/offline_sync_demo/lib/src/offline_replica.dart

@@ -2,8 +2,6 @@ import 'dart:io';
 
 import 'package:flutter/foundation.dart' show debugPrint, kDebugMode;
 import 'package:path/path.dart' as p;
-import 'package:serverpod_database/serverpod_database.dart'
-    show ClientDatabaseSession;
 import 'package:serverpod_offline_sync_client/serverpod_offline_sync_client.dart';
 import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
 
@@ -13,8 +11,8 @@ import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_c
 /// This class is the entire surface a real integration needs — read it first to
 /// understand the package. The four steps are:
 ///
-///   1. [open] opens a local SQLite database ([Client.createSession])
+///   1. [open] uses [Client.createSyncSession] to open and initialize a local
-///      and wraps it in a [OfflineSyncDatabaseSession]. Use
+///      SQLite database as an [OfflineSyncDatabaseSession]. Use
 ///      [openOrReset] when the file may predate the shipped
 ///      schema. The rest of the app reads and writes generated models through
 ///      [session].
@@ -23,16 +21,13 @@ import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_c
 ///      e.g. `Person.db.insertRow(offlineSyncSession, person)`).
 ///   3. [syncOnce] / [syncContinuously] push local changes and merge remote
 ///      ones through a Serverpod [Client].
-///   4. [reset] / [close] tear the replica down.
+///   4. [reset] clears local data; [close] releases the database connection.
 ///
 /// It deliberately holds **no UI state**: the demo's `DemoController` layers
 /// busy/error/projection bookkeeping on top of these calls so that this file
 /// stays a clean reference for the package itself.
 class OfflineReplica {
-  OfflineReplica._(this._rawSession, this.session, this.persistentUserId);
+  OfflineReplica._(this.session, this.persistentUserId);
-
-  /// Owns the underlying SQLite connection; closed by [close].
-  final ClientDatabaseSession _rawSession;
 
   /// CRDT-aware session used for every model read and write. By default it only
   /// exposes visible (non-tombstoned) rows; queries opt into hidden rows with
@@ -44,26 +39,19 @@ class OfflineReplica {
 
   /// Opens [databasePath] through [client] and wraps it for CRDT sync.
   ///
-  /// [Client.createSession] opens the local client database (running
+  /// [Client.createSyncSession] runs client migrations, wraps the database for
-  /// client migrations); [OfflineSyncDatabaseSession.wraps] layers CRDT
+  /// the generated sync tables, and initializes CRDT tracking for this user.
-  /// tracking over it; and `db.initialize()` establishes this device's CRDT
-  /// node so it can take part in sync.
   static Future<OfflineReplica> open({
     required Client client,
     required String databasePath,
     required UuidValue persistentUserId,
   }) async {
-    final rawSession = await client.createSession(
+    final session = await client.createSyncSession(
       databasePath,
       isDebugMode: kDebugMode,
-    );
-    final offlineSyncSession = OfflineSyncDatabaseSession.wraps(
-      rawSession,
-      syncTables: syncTables,
       persistentUserId: persistentUserId,
     );
-    await offlineSyncSession.db.initialize();
+    return OfflineReplica._(session, persistentUserId);
-    return OfflineReplica._(rawSession, offlineSyncSession, persistentUserId);
   }
 
   /// Like [open], but deletes [databasePath] and retries once when opening
@@ -160,5 +148,5 @@ class OfflineReplica {
   }
 
   /// Closes the underlying database connection.
-  Future<void> close() => _rawSession.close();
+  Future<void> close() => session.close();
 }

packages/serverpod_offline_sync/example/README.md

@@ -0,0 +1,12 @@
+# Example
+
+The [Offline Sync Demo](https://github.com/marcelomendoncasoares/serverpod_offline_sync/tree/main/example/offline_sync_demo)
+is the shared example app for `serverpod_offline_sync`,
+`serverpod_offline_sync_client`, and `serverpod_offline_sync_server`.
+
+The desktop Flutter app shows two independent local replicas alongside the
+server's merged data, so you can edit offline, sync each replica, and watch them
+converge. See the demo's README for setup and running instructions.
+
+For the core client integration, start with
+[`offline_replica.dart`](https://github.com/marcelomendoncasoares/serverpod_offline_sync/blob/main/example/offline_sync_demo/lib/src/offline_replica.dart).

packages/serverpod_offline_sync/lib/src/database/session.dart

@@ -66,9 +66,32 @@ class OfflineSyncDatabaseSession implements DatabaseSession {
     syncBatchSize: syncBatchSize,
     continuousSyncInterval: continuousSyncInterval,
     persistentUserId: persistentUserId,
-  );
+  ).._wrappedSession = session;
 
   final OfflineSyncDatabase _db;
+  DatabaseSession? _wrappedSession;
+  Future<void>? _closeFuture;
+
+  /// Closes the underlying client database.
+  ///
+  /// Supported for sessions created by [OfflineSyncDatabaseSession.wraps] around
+  /// a [ClientDatabaseSession], including nested sync-session wrappers and the
+  /// sessions returned by the generated client's `createSyncSession` method.
+  /// Repeated or concurrent calls share the same close operation.
+  ///
+  /// Throws [UnsupportedError] for sessions constructed directly from a
+  /// [Database] or wrapping a non-client session. Those connections must be
+  /// closed through their owner, such as the server.
+  Future<void> close() => _closeFuture ??= switch (_wrappedSession) {
+    final ClientDatabaseSession session => session.close(),
+    final OfflineSyncDatabaseSession session => session.close(),
+    _ => Future<void>.error(
+      UnsupportedError(
+        'Only sync sessions wrapping a ClientDatabaseSession can be closed. '
+        'Close the underlying database through its owner.',
+      ),
+    ),
+  };
 
   @override
   OfflineSyncDatabase get db => _db;

packages/serverpod_offline_sync_client/example/README.md

@@ -0,0 +1 @@
+../../serverpod_offline_sync/example/README.md

packages/serverpod_offline_sync_server/example/README.md

@@ -0,0 +1 @@
+../../serverpod_offline_sync/example/README.md

test/serverpod_offline_sync_test_server/test/integration/database/sync_session_lifecycle_test.dart

@@ -0,0 +1,175 @@
+import 'dart:io';
+
+import 'package:path/path.dart' as p;
+import 'package:serverpod_database/serverpod_database.dart' show ClientDatabaseSession;
+import 'package:serverpod_offline_sync_client/serverpod_offline_sync_client.dart';
+import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
+import 'package:test/test.dart';
+
+void main() {
+  late Directory directory;
+  late Client client;
+  final userId = UuidValue.fromString('00000000-0000-4000-8000-000000000001');
+
+  setUpAll(() async {
+    directory = await Directory.systemTemp.createTemp('sync_session_lifecycle_');
+    client = Client('http://localhost:8080/');
+  });
+
+  tearDownAll(() async {
+    client.close();
+    await directory.delete(recursive: true);
+  });
+
+  group('Given a generated sync session with a persisted Person,', () {
+    late String path;
+    late OfflineSyncDatabaseSession session;
+    late Person person;
+    late UuidValue nodeId;
+
+    setUpAll(() async {
+      path = p.join(directory.path, 'reopened.db');
+      session = await client.createSyncSession(path, persistentUserId: userId);
+      addTearDown(session.close);
+      person = await Person.db.insertRow(session, Person(name: 'Alice'));
+      nodeId = await session.db.currentNodeId();
+    });
+
+    group('when it is closed and the database is reopened,', () {
+      late OfflineSyncDatabaseSession reopened;
+
+      setUpAll(() async {
+        await session.close();
+        reopened = await client.createSyncSession(path, persistentUserId: userId);
+        addTearDown(reopened.close);
+      });
+
+      test('then the closed session rejects database queries.', () async {
+        await expectLater(session.db.testConnection(), throwsStateError);
+      });
+
+      test('then the stored Person is available in the reopened session.', () async {
+        final loaded = await Person.db.findById(reopened, person.id!);
+
+        expect(loaded?.name, 'Alice');
+        expect(loaded?.spaceId, person.spaceId);
+      });
+
+      test('then the replica keeps its CRDT node identity.', () async {
+        expect(await reopened.db.currentNodeId(), nodeId);
+      });
+    });
+  });
+
+  group('Given a generated sync session,', () {
+    late OfflineSyncDatabaseSession session;
+
+    setUpAll(() async {
+      session = await client.createSyncSession(
+        p.join(directory.path, 'repeated.db'),
+        persistentUserId: userId,
+      );
+      addTearDown(session.close);
+    });
+
+    group('when it is closed concurrently and closed again afterwards,', () {
+      setUpAll(() async {
+        await Future.wait([session.close(), session.close()]);
+        await session.close();
+      });
+
+      test('then every close completes and the database stays closed.', () async {
+        await expectLater(session.db.testConnection(), throwsStateError);
+      });
+    });
+  });
+
+  group('Given two generated sync sessions for different replica files,', () {
+    late OfflineSyncDatabaseSession first;
+    late OfflineSyncDatabaseSession second;
+
+    setUpAll(() async {
+      first = await client.createSyncSession(
+        p.join(directory.path, 'first.db'),
+        persistentUserId: userId,
+      );
+      addTearDown(first.close);
+      second = await client.createSyncSession(
+        p.join(directory.path, 'second.db'),
+        persistentUserId: userId,
+      );
+      addTearDown(second.close);
+    });
+
+    group('when one session is closed and a Person is written to the other,', () {
+      late Person person;
+
+      setUpAll(() async {
+        await first.close();
+        person = await Person.db.insertRow(second, Person(name: 'Bob'));
+      });
+
+      test('then the other replica can still read the persisted Person.', () async {
+        expect((await Person.db.findById(second, person.id!))?.name, 'Bob');
+      });
+    });
+  });
+
+  group('Given a generated sync session wrapped in another sync session,', () {
+    late OfflineSyncDatabaseSession session;
+    late OfflineSyncDatabaseSession outer;
+
+    setUpAll(() async {
+      session = await client.createSyncSession(
+        p.join(directory.path, 'nested.db'),
+        persistentUserId: userId,
+      );
+      addTearDown(session.close);
+      outer = OfflineSyncDatabaseSession.wraps(session, syncTables: syncTables);
+    });
+
+    group('when the outer session is closed,', () {
+      setUpAll(() async {
+        await outer.close();
+      });
+
+      test('then the underlying client database is closed.', () async {
+        await expectLater(session.db.testConnection(), throwsStateError);
+      });
+    });
+  });
+
+  group(
+    'Given a sync session constructed directly from an externally owned database,',
+    () {
+      late ClientDatabaseSession owner;
+      late OfflineSyncDatabaseSession session;
+
+      setUpAll(() async {
+        owner = await client.createSession(p.join(directory.path, 'external.db'));
+        addTearDown(owner.close);
+        session = OfflineSyncDatabaseSession(owner.db, syncTables: syncTables);
+      });
+
+      group('when closing the sync session is attempted,', () {
+        Object? closeError;
+
+        setUpAll(() async {
+          try {
+            await session.close();
+          } on Object catch (error) {
+            closeError = error;
+          }
+        });
+
+        test(
+          'then it rejects the operation and leaves the owner connection open.',
+          () async {
+            expect(closeError, isUnsupportedError);
+            expect(await owner.db.testConnection(), isTrue);
+          },
+        );
+      });
+    },
+  );
+}

test/serverpod_offline_sync_test_server/test/integration/server/offline_sync_database_interceptor_test.dart

@@ -40,6 +40,21 @@ void main() {
           expect(session.offlineSync, isNotNull);
         },
       );
+
+      test(
+        'when closing a sync wrapper around a server session is attempted, '
+        'then it rejects the operation and leaves the server connection open.',
+        () async {
+          final wrapped = OfflineSyncDatabaseSession.wraps(
+            session,
+            syncTables: syncTables,
+          );
+
+          await expectLater(wrapped.close(), throwsUnsupportedError);
+
+          expect(await session.db.testConnection(), isTrue);
+        },
+      );
     },
   );