fix: Make the CrdtDatabase class ephemeral (#82)

packages/serverpod_offline_sync_client/lib/src/crdt/sync.dart

@@ -5,6 +5,7 @@ import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart
 import 'package:uuid/uuid.dart';
 
 import '../database/database.dart';
+import '../database/recorder.dart';
 import '../managers/scope.dart';
 import '../protocol/protocol.dart';
 import '../utils/case_when.dart' show Case;
@@ -32,6 +33,9 @@ class CrdtSync {
     /// The serialization manager to use for deserializing merge changes.
     required DatabaseSerializationManager serializationManager,
 
+    /// Shared CRDT database metadata.
+    CrdtDatabaseContext? databaseContext,
+
     /// Maximum number of merge changes sent in one sync stream message.
     int syncBatchSize = defaultSyncBatchSize,
 
@@ -39,6 +43,12 @@ class CrdtSync {
     Duration continuousSyncInterval = defaultContinuousSyncInterval,
   }) : _syncTables = syncTables,
        _serializationManager = serializationManager,
+       _databaseContext =
+           databaseContext ??
+           CrdtDatabaseContext(
+             syncTables: syncTables,
+             serializationManager: serializationManager,
+           ),
        _syncBatchSize = syncBatchSize,
        _continuousSyncInterval = continuousSyncInterval {
     if (syncBatchSize < 1) {
@@ -54,38 +64,20 @@ class CrdtSync {
 
   final List<Table> _syncTables;
   final DatabaseSerializationManager _serializationManager;
+  final CrdtDatabaseContext _databaseContext;
   final int _syncBatchSize;
   final Duration _continuousSyncInterval;
 
-  static CrdtSync? _instance;
+  /// Wraps [database] in a CRDT-aware database using this sync context.
-
+  CrdtDatabase wrapDatabase(Database database, {UuidValue? persistentUserId}) {
-  /// The singleton instance of [CrdtSync]. Throws a [StateError] if it is not
+    if (database is CrdtDatabase) return database;
-  /// initialized.
+    return CrdtDatabase(
-  static CrdtSync get instance =>
+      database,
-      _instance ??
+      syncTables: _syncTables,
-      (throw StateError(
+      syncBatchSize: _syncBatchSize,
-        'The CrdtSync has not been initialized. Call pod.initializeCrdtSync(...) '
+      continuousSyncInterval: _continuousSyncInterval,
-        'during server startup to configure the CRDT sync.',
+      persistentUserId: persistentUserId,
-      ));
+      context: _databaseContext,
-
-  /// Configures the shared singleton used by server endpoints.
-  ///
-  /// Use [syncBatchSize] to control the maximum number of merge changes carried
-  /// by each [CrdtSyncMergeChunk] stream event.
-  ///
-  /// The [continuousSyncInterval] controls how long a continuous sync session
-  /// waits after completing one sync round before checking for local changes.
-  static void initialize({
-    required List<Table> syncTables,
-    required DatabaseSerializationManager serializationManager,
-    int syncBatchSize = defaultSyncBatchSize,
-    Duration continuousSyncInterval = defaultContinuousSyncInterval,
-  }) {
-    _instance = CrdtSync(
-      syncTables: syncTables,
-      serializationManager: serializationManager,
-      syncBatchSize: syncBatchSize,
-      continuousSyncInterval: continuousSyncInterval,
     );
   }
 
@@ -208,7 +200,7 @@ class CrdtSync {
   }) async {
     if (mergeSet.isEmpty) return null;
     final maxSyncedHlc = mergeSet.maxHlc;
-    final crdtDb = await _openCrdtDatabase(session);
+    final crdtDb = _openCrdtDatabase(session);
     await crdtDb.mergeChanges(mergeSet, scopeId: userId);
     if (maxSyncedHlc != null) {
       await crdtDb.recordSyncCheckpoint(otherNodeId, maxSyncedHlc, userId: userId);
@@ -398,15 +390,12 @@ class CrdtSync {
     }
   }
 
-  Future<CrdtDatabase> _openCrdtDatabase(DatabaseSession session) async {
+  CrdtDatabase _openCrdtDatabase(DatabaseSession session) {
     final db = session.db;
-    if (db is CrdtDatabase) {
+    // The wrapper is ephemeral and every operation performed on it lazily
-      return db;
+    // ensures initialization, so there is nothing to eagerly initialize here.
-    }
+    // Calling `initialize()` would re-run the per-session setup.
-
+    return db is CrdtDatabase ? db : wrapDatabase(db);
-    final crdtDb = CrdtDatabase(db, syncTables: _syncTables);
-    await crdtDb.initialize();
-    return crdtDb;
   }
 
   Stream<CrdtMergeChange> _streamPendingChanges(

packages/serverpod_offline_sync_client/lib/src/database/database.dart

@@ -25,11 +25,14 @@ final scopeForTransaction = <Transaction, CrdtScope>{};
 class CrdtDatabase implements Database {
   /// Creates a CRDT-aware database wrapper around the inner database.
   CrdtDatabase(
-    this._delegate, {
+    Database delegate, {
 
     /// The list of tables to sync with CRDT.
     required List<Table> syncTables,
 
+    /// Shared CRDT database metadata.
+    CrdtDatabaseContext? context,
+
     /// Maximum number of merge changes sent in one sync stream message.
     int syncBatchSize = CrdtSync.defaultSyncBatchSize,
 
@@ -40,54 +43,58 @@ class CrdtDatabase implements Database {
     /// databases operating on the client side, where all data is for the same user.
     /// Otherwise, the user ID must be passed through the transaction.
     UuidValue? persistentUserId,
+  }) : this._(
+         delegate,
+         context ??
+             CrdtDatabaseContext(
+               syncTables: syncTables,
+               serializationManager: delegate.serializationManager,
+             ),
+         syncTables: syncTables,
+         syncBatchSize: syncBatchSize,
+         continuousSyncInterval: continuousSyncInterval,
+         persistentUserId: persistentUserId,
+       );
+
+  CrdtDatabase._(
+    this._delegate,
+    this._context, {
+    required List<Table> syncTables,
+    required int syncBatchSize,
+    required Duration continuousSyncInterval,
+    required UuidValue? persistentUserId,
   }) : _syncTables = syncTables,
        _syncBatchSize = syncBatchSize,
        _continuousSyncInterval = continuousSyncInterval,
        _recorder = CrdtMutationRecorder(
          _delegate,
+         context: _context,
          persistentUserId: persistentUserId,
-         syncTables: syncTables,
        );
 
   final Database _delegate;
+  final CrdtDatabaseContext _context;
   final List<Table> _syncTables;
   final int _syncBatchSize;
   final Duration _continuousSyncInterval;
 
   final CrdtMutationRecorder _recorder;
-  bool _initialized = false;
-  Future<void>? _initializing;
 
-  late final CrdtSync _sync = CrdtSync(
+  late final _sync = CrdtSync(
     syncTables: _syncTables,
     serializationManager: serializationManager,
     syncBatchSize: _syncBatchSize,
     continuousSyncInterval: _continuousSyncInterval,
+    databaseContext: _context,
   );
 
   /// Initializes the CRDT database.
   Future<void> initialize() async {
-    final initializing = _recorder.initialize().then((_) {
+    await _recorder.initialize();
-      _initialized = true;
-    });
-    _initializing = initializing;
-    try {
-      await initializing;
-    } finally {
-      if (identical(_initializing, initializing)) {
-        _initializing = null;
-      }
-    }
   }
 
   Future<void> _ensureInitialized() async {
-    if (_initialized) return;
+    await _recorder.ensureInitialized();
-    final initializing = _initializing;
-    if (initializing != null) {
-      await initializing;
-      return;
-    }
-    await initialize();
   }
 
   /// The hash describing the synchronized schema configured for this database.
@@ -342,6 +349,7 @@ class CrdtDatabase implements Database {
     Expression? updateWhere,
     Transaction? transaction,
   }) async {
+    await _ensureInitialized();
     // TODO: Implement CRDT upsert.
     throw UnimplementedError('CRDT upsert is not implemented.');
   }

packages/serverpod_offline_sync_client/lib/src/database/merge.dart

@@ -828,7 +828,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
     String tableName,
     Map<String, Object?> data,
   ) {
-    final columns = _columnsByTableAndName[tableName];
+    final columns = _context._columnsByTableAndName[tableName];
     if (columns == null) return {};
 
     return {

packages/serverpod_offline_sync_client/lib/src/database/merge_utils/foreign_key_projector.dart

@@ -32,6 +32,12 @@ typedef _ForeignKeyDefaultProjection = ({bool valid, UuidValue? value});
 typedef _DomainRowUpdatesByKey = Map<_MergeRowKey, Map<String, Object?>>;
 
 extension _CrdtForeignKeyProjector on CrdtMutationRecorder {
+  List<_ForeignKeyEdge> get _foreignKeyEdges => _context._foreignKeyEdges;
+  Map<String, List<_ForeignKeyEdge>> get _foreignKeyEdgesByParentTable =>
+      _context._foreignKeyEdgesByParentTable;
+  Map<String, List<_ForeignKeyEdge>> get _foreignKeyEdgesByChildTable =>
+      _context._foreignKeyEdgesByChildTable;
+
   bool _mergeOperationsMayAffectForeignKeys(
     List<CrdtMergeChange> operations,
   ) {

packages/serverpod_offline_sync_client/lib/src/database/recorder.dart

@@ -58,38 +58,96 @@ enum _ForeignKeyTargetPresence {
   hidden,
 }
 
-/// Persists additional CRDT rows after a mutating ORM operation completes.
+typedef _CrdtSchema = Map<String, (int, Map<String, CrdtSchemaColumn>)>;
+
+/// Process-level CRDT database metadata shared by ephemeral database wrappers.
 ///
-/// Callbacks receive the underlying database (not the CRDT proxy) and the
+/// A single context is created per `CrdtSync` (i.e. once per Serverpod
-/// active transaction. Use that database for follow-up inserts so work is not
+/// instance) and shared by every ephemeral [CrdtDatabase], so the schema is
-/// wrapped again by the proxy.
+/// synchronized once per process rather than once per `Session`.
-class CrdtMutationRecorder {
+///
-  /// Creates a [CrdtMutationRecorder] instance.
+/// The cached schema holds database-assigned identifiers and is never
-  CrdtMutationRecorder(
+/// invalidated for the lifetime of the context. This assumes the CRDT schema
-    this._db, {
+/// rows stay stable while the process is alive, which holds for a normal server
-    required this.persistentUserId,
+/// whose database is not reset underneath it. If the schema rows are dropped and
+/// re-created with different identifiers while the process lives, a new context
+/// must be created.
+class CrdtDatabaseContext {
+  /// Creates a [CrdtDatabaseContext] for the configured synchronized tables.
+  CrdtDatabaseContext({
     required this.syncTables,
-  }) : assert(
+    required DatabaseSerializationManager serializationManager,
-         _db is! CrdtDatabase,
+  }) : _tableDefinitions = serializationManager.getTargetTableDefinitions();
-         'The database must be the user database, not the CRDT database. '
-         'Passing a CRDT database would cause an infinite recursion.',
-       );
 
-  final Database _db;
+  /// The list of tables to sync with CRDT.
+  final List<Table> syncTables;
 
-  late final DatabaseSession _dbSession = _db.session;
+  final List<TableDefinition> _tableDefinitions;
 
-  late final _session = BasicDatabaseSession(_db);
+  _CrdtSchema? _schema;
+  Future<_CrdtSchema>? _schemaFuture;
 
-  late final CrdtScopeManager _scopeManager = CrdtScopeManager(_dbSession);
+  /// Initializes shared schema rows and caches their generated identifiers.
+  Future<void> initialize(DatabaseSession session) async {
+    if (_schema != null) return;
 
-  final Map<UuidValue, HlcManager> _hlcManagers = {};
+    final schemaFuture = _schemaFuture ??= _loadSchema(session);
+    try {
+      _schema = await schemaFuture;
+    } catch (_) {
+      if (identical(_schemaFuture, schemaFuture)) {
+        _schemaFuture = null;
+      }
+      rethrow;
+    }
+  }
 
-  late Map<String, (int, Map<String, CrdtSchemaColumn>)> _schema;
+  Future<_CrdtSchema> _loadSchema(DatabaseSession session) async {
+    final schemaRegistry = CrdtSchemaRegistry(
+      session,
+      syncTables: syncTables,
+      tableDefinitions: _tableDefinitions,
+    );
+    final (tableRows, columnRows) = await schemaRegistry.syncAndGetSchema();
 
-  late final _tableDefinitionsByName = {
+    final columnsByTableId = <int, Map<String, CrdtSchemaColumn>>{};
-    for (final table in _db.serializationManager.getTargetTableDefinitions())
+    for (final column in columnRows) {
-      table.name: table,
+      columnsByTableId.putIfAbsent(column.tblId, () => {})[column.name] = column;
+    }
+
+    return {
+      for (final t in tableRows) t.name: (t.id!, columnsByTableId[t.id!] ?? {}),
+    };
+  }
+
+  _CrdtSchema get _schemaSnapshot =>
+      _schema ??
+      (throw StateError(
+        'The CRDT database has not been initialized. Call '
+        'CrdtDatabase.initialize() before using CRDT database operations.',
+      ));
+
+  /// Returns the local [CrdtSchemaTable] id for [tableName], or null when the
+  /// table is not registered for CRDT synchronization.
+  int? _tableIdForName(String tableName) => _schemaSnapshot[tableName]?.$1;
+
+  late final Map<String, Table> _syncTableByName = {
+    for (final t in syncTables) t.tableName: t,
+  };
+
+  late final Map<String, TableDefinition> _tableDefinitionsByName = {
+    for (final table in _tableDefinitions) table.name: table,
+  };
+
+  late final Map<String, Map<String, ColumnDefinition>> _columnsByTableAndName = {
+    for (final table in _tableDefinitionsByName.values)
+      table.name: {for (final column in table.columns) column.name: column},
+  };
+
+  late final Map<String, Set<String>> _syncedTableColumnNamesForMerge = {
+    for (final MapEntry(key: k, value: cols) in _columnsByTableAndName.entries)
+      if (_syncTableByName.containsKey(k))
+        k: cols.keys.where((columnName) => columnName != 'scopeId').toSet(),
   };
 
   late final Map<String, List<_ReferencingForeignKey>> _foreignKeysByReferencedTable =
@@ -157,38 +215,94 @@ class CrdtMutationRecorder {
 
   final _uniqueIndexesByTableName = <String, List<_UniqueIndexConflictRelease>>{};
 
-  late final Map<String, Map<String, ColumnDefinition>> _columnsByTableAndName = {
+  bool _isCrdtTrackedTableName(String tableName) {
-    for (final table in _tableDefinitionsByName.values)
+    return _syncTableByName.containsKey(tableName);
-      table.name: {for (final column in table.columns) column.name: column},
+  }
-  };
+
+  Object? _defaultValueForColumn(String tableName, String columnName) {
+    final column = _columnsByTableAndName[tableName]?[columnName];
+    if (column == null) return null;
+    final defaultValue = column.columnDefault;
+    if (defaultValue == null) return null;
+    if (column.columnType == ColumnType.uuid) {
+      final unquoted = defaultValue.replaceAll("'", '');
+      if (unquoted == 'random' || unquoted == 'random_v7') return null;
+      return UuidValue.withValidation(unquoted);
+    }
+    return defaultValue.replaceAll("'", '');
+  }
+}
+
+/// Persists additional CRDT rows after a mutating ORM operation completes.
+///
+/// Callbacks receive the underlying database (not the CRDT proxy) and the
+/// active transaction. Use that database for follow-up inserts so work is not
+/// wrapped again by the proxy.
+class CrdtMutationRecorder {
+  /// Creates a [CrdtMutationRecorder] instance.
+  CrdtMutationRecorder(
+    this._db, {
+    required CrdtDatabaseContext context,
+    required this.persistentUserId,
+  }) : _context = context,
+       assert(
+         _db is! CrdtDatabase,
+         'The database must be the user database, not the CRDT database. '
+         'Passing a CRDT database would cause an infinite recursion.',
+       );
+
+  final Database _db;
+  final CrdtDatabaseContext _context;
+
+  /// The raw database session used for CRDT metadata reads and writes.
+  late final _session = _db.session;
+
+  late final _scopeManager = CrdtScopeManager(_session);
+
+  final Map<UuidValue, HlcManager> _hlcManagers = {};
+  Future<void>? _ensureInitializedFuture;
+  var _isInitialized = false;
+
+  _CrdtSchema get _schema => _context._schemaSnapshot;
+  Map<String, TableDefinition> get _tableDefinitionsByName =>
+      _context._tableDefinitionsByName;
 
   /// Initializes the CRDT recorder.
   ///
-  /// Safe to call again after the database was wiped (e.g. test `tearDown`)
+  /// Clears this recorder's live scope/HLC caches, then ensures the shared
-  /// for in-memory schema ids to match new rows.
+  /// [CrdtDatabaseContext] metadata is loaded. The shared schema is loaded only
+  /// once per process and is not reloaded here; see [CrdtDatabaseContext] for
+  /// the cache lifetime assumptions.
   Future<void> initialize() async {
     _scopeManager.clearCache();
     _hlcManagers.clear();
+    _ensureInitializedFuture = null;
+    _isInitialized = false;
 
+    await ensureInitialized();
+  }
+
+  /// Ensures the recorder is ready without clearing live per-session caches.
+  Future<void> ensureInitialized() async {
+    if (_isInitialized) return;
+
+    final initializeFuture = _ensureInitializedFuture ??= _initializeOnce();
+    try {
+      await initializeFuture;
+    } catch (_) {
+      if (identical(_ensureInitializedFuture, initializeFuture)) {
+        _ensureInitializedFuture = null;
+      }
+      rethrow;
+    }
+  }
+
+  Future<void> _initializeOnce() async {
+    await _context.initialize(_session);
     if (persistentUserId != null) {
       await _scopeManager.getOrCreate(persistentUserId!);
     }
-
+    _isInitialized = true;
-    final schemaRegistry = CrdtSchemaRegistry(_dbSession, syncTables: syncTables);
-    final (tableRows, columnRows) = await schemaRegistry.syncAndGetSchema();
-
-    final columnsByTableId = <int, Map<String, CrdtSchemaColumn>>{};
-    for (final column in columnRows) {
-      columnsByTableId.putIfAbsent(column.tblId, () => {})[column.name] = column;
-    }
-
-    _schema = {
-      for (final t in tableRows)
-        t.name: (
-          t.id!,
-          columnsByTableId[t.id!] ?? {},
-        ),
-    };
   }
 
   /// The user ID to use for all CRDT operations. This should only be used for
@@ -197,17 +311,12 @@ class CrdtMutationRecorder {
   final UuidValue? persistentUserId;
 
   /// The list of tables to sync with CRDT.
-  final List<Table> syncTables;
+  List<Table> get syncTables => _context.syncTables;
 
-  late final _syncTableByName = {
+  Map<String, Table> get _syncTableByName => _context._syncTableByName;
-    for (final t in syncTables) t.tableName: t,
-  };
 
-  late final Map<String, Set<String>> _syncedTableColumnNamesForMerge = {
+  Map<String, Set<String>> get _syncedTableColumnNamesForMerge =>
-    for (final MapEntry(key: k, value: cols) in _columnsByTableAndName.entries)
+      _context._syncedTableColumnNamesForMerge;
-      if (_syncTableByName.containsKey(k))
-        k: cols.keys.where((columnName) => columnName != 'scopeId').toSet(),
-  };
 
   /// Whether the given table is tracked by CRDT.
   bool isCrdtTracked<T extends TableRow>([Table? table]) {
@@ -218,12 +327,12 @@ class CrdtMutationRecorder {
 
   /// Whether the given table name is tracked by CRDT.
   bool _isCrdtTrackedTableName(String tableName) {
-    return _syncTableByName.containsKey(tableName);
+    return _context._isCrdtTrackedTableName(tableName);
   }
 
   /// Returns the local [CrdtSchemaTable] id for [tableName], or null when the
   /// table is not registered for CRDT synchronization.
-  int? tableIdForName(String tableName) => _schema[tableName]?.$1;
+  int? tableIdForName(String tableName) => _context._tableIdForName(tableName);
 
   /// Returns the user scoping CRDT visibility for queries, or null when no
   /// user is associated with [transaction] and no persistent user exists.
@@ -919,7 +1028,7 @@ WHERE c."id" IN ($whereRowIds)
 
     final cascadeDeletes = <String, Set<UuidValue>>{};
     final foreignKeys =
-        _foreignKeysByReferencedTable[parentTableName] ??
+        _context._foreignKeysByReferencedTable[parentTableName] ??
         const <_ReferencingForeignKey>[];
     for (final reference in foreignKeys) {
       final parentValuesById = reference.parentColumn == 'id'
@@ -964,7 +1073,7 @@ WHERE c."id" IN ($whereRowIds)
               transaction,
             );
           case ForeignKeyAction.setDefault:
-            final defaultValue = _defaultValueForColumn(
+            final defaultValue = _context._defaultValueForColumn(
               reference.childTableName,
               reference.childColumn,
             );
@@ -1049,7 +1158,7 @@ WHERE c."id" IN ($whereRowIds)
   List<_UniqueIndexConflictRelease> _uniqueIndexesForTable(
     TableDefinition tableDefinition,
   ) {
-    return _uniqueIndexesByTableName.putIfAbsent(
+    return _context._uniqueIndexesByTableName.putIfAbsent(
       tableDefinition.name,
       () => _syncableUniqueIndexesForTable(
         tableDefinition,
@@ -1176,19 +1285,6 @@ WHERE "id" IN (${_sqlLiteralList(rowIds)})
     };
   }
 
-  Object? _defaultValueForColumn(String tableName, String columnName) {
-    final column = _columnsByTableAndName[tableName]?[columnName];
-    if (column == null) return null;
-    final defaultValue = column.columnDefault;
-    if (defaultValue == null) return null;
-    if (column.columnType == ColumnType.uuid) {
-      final unquoted = defaultValue.replaceAll("'", '');
-      if (unquoted == 'random' || unquoted == 'random_v7') return null;
-      return UuidValue.withValidation(unquoted);
-    }
-    return defaultValue.replaceAll("'", '');
-  }
-
   Object? _conflictFreeValue(
     _UniqueColumnConflictRelease column,
     Object? value,

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

@@ -4,6 +4,7 @@ import 'package:uuid/uuid.dart';
 
 import '../crdt/sync.dart';
 import 'database.dart';
+import 'recorder.dart';
 
 /// Wraps a [DatabaseSession] to provide a [CrdtDatabase] as [DatabaseSession.db].
 class CrdtDatabaseSession implements DatabaseSession {
@@ -14,6 +15,9 @@ class CrdtDatabaseSession implements DatabaseSession {
     /// The list of tables to sync with CRDT.
     required List<Table> syncTables,
 
+    /// Shared CRDT database metadata.
+    CrdtDatabaseContext? context,
+
     /// Maximum number of merge changes sent in one sync stream message.
     int syncBatchSize = CrdtSync.defaultSyncBatchSize,
 
@@ -24,13 +28,16 @@ class CrdtDatabaseSession implements DatabaseSession {
     /// databases operating on the client side, where all data is for the same user.
     /// Otherwise, the user ID must be passed through the transaction.
     UuidValue? persistentUserId,
-  }) : _db = CrdtDatabase(
+  }) : _db = db is CrdtDatabase
-         db,
+           ? db
-         syncTables: syncTables,
+           : CrdtDatabase(
-         syncBatchSize: syncBatchSize,
+               db,
-         continuousSyncInterval: continuousSyncInterval,
+               syncTables: syncTables,
-         persistentUserId: persistentUserId,
+               context: context,
-       );
+               syncBatchSize: syncBatchSize,
+               continuousSyncInterval: continuousSyncInterval,
+               persistentUserId: persistentUserId,
+             );
 
   /// Creates a [CrdtDatabaseSession] instance that wraps a [DatabaseSession].
   factory CrdtDatabaseSession.wraps(
@@ -39,6 +46,9 @@ class CrdtDatabaseSession implements DatabaseSession {
     /// The list of tables to sync with CRDT.
     required List<Table> syncTables,
 
+    /// Shared CRDT database metadata.
+    CrdtDatabaseContext? context,
+
     /// Maximum number of merge changes sent in one sync stream message.
     int syncBatchSize = CrdtSync.defaultSyncBatchSize,
 
@@ -52,6 +62,7 @@ class CrdtDatabaseSession implements DatabaseSession {
   }) => CrdtDatabaseSession(
     session.db,
     syncTables: syncTables,
+    context: context,
     syncBatchSize: syncBatchSize,
     continuousSyncInterval: continuousSyncInterval,
     persistentUserId: persistentUserId,

packages/serverpod_offline_sync_server/lib/src/business/crdt_sync.dart

@@ -1,12 +1,34 @@
 import 'package:serverpod/serverpod.dart';
 import 'package:serverpod_offline_sync_client/serverpod_offline_sync_client.dart';
 
+/// The CRDT sync configured per [Serverpod] instance.
+///
+/// Keyed by the [Serverpod] instance so each pod owns its own [CrdtSync] (and
+/// the [CrdtDatabaseContext] it carries) instead of sharing a single
+/// process-wide singleton.
+final _crdtSyncByServerpod = Expando<CrdtSync>('crdtSync');
+
+/// Intercepts each Serverpod session database with a CRDT-aware database once
+/// [CrdtSyncInitialize.initializeCrdtSync] has configured sync.
+///
+/// When sync has not been configured for the session's [Serverpod], the
+/// original [inner] database is returned unchanged.
+Database crdtDatabaseInterceptor(Session session, Database inner) {
+  final crdtSync = _crdtSyncByServerpod[session.server.serverpod];
+  return crdtSync?.wrapDatabase(inner) ?? inner;
+}
+
 /// Extension methods for [Serverpod] to configure the CRDT sync on the server.
 extension CrdtSyncInitialize on Serverpod {
-  /// Configures the endpoint with the given sync tables.
+  /// Configures the CRDT sync with the given sync tables.
   ///
   /// Must be called during server startup before any sync requests are made.
-  /// Will override any previous initialization.
+  /// Will override any previous initialization for this [Serverpod] instance.
+  ///
+  /// The [Serverpod] instance must be constructed with [crdtDatabaseInterceptor]
+  /// as its `databaseInterceptor`. Otherwise each session's [Session.db] stays a
+  /// plain database and server-side ORM mutations on synced tables are not
+  /// CRDT-tracked.
   ///
   /// [syncBatchSize] controls the maximum number of merge changes carried by
   /// each sync stream chunk.
@@ -18,7 +40,7 @@ extension CrdtSyncInitialize on Serverpod {
     int syncBatchSize = CrdtSync.defaultSyncBatchSize,
     Duration continuousSyncInterval = CrdtSync.defaultContinuousSyncInterval,
   }) {
-    CrdtSync.initialize(
+    _crdtSyncByServerpod[this] = CrdtSync(
       syncTables: syncTables,
       serializationManager: serializationManager,
       syncBatchSize: syncBatchSize,
@@ -26,3 +48,15 @@ extension CrdtSyncInitialize on Serverpod {
     );
   }
 }
+
+/// Extension to access the CRDT sync for [Session] from the [Serverpod] instance.
+extension CrdtSessionExtension on Session {
+  /// Returns the CRDT sync configured for this session.
+  CrdtSync get crdt =>
+      _crdtSyncByServerpod[server.serverpod] ??
+      (throw StateError(
+        'The CrdtSync has not been initialized for this Serverpod instance. '
+        'Call pod.initializeCrdtSync(...) during server startup to configure '
+        'the CRDT sync.',
+      ));
+}

packages/serverpod_offline_sync_server/lib/src/endpoints/crdt_sync_endpoint.dart

@@ -3,6 +3,8 @@ import 'dart:async';
 import 'package:serverpod/serverpod.dart';
 import 'package:serverpod_offline_sync_client/serverpod_offline_sync_client.dart';
 
+import '../business/crdt_sync.dart';
+
 /// Endpoint for CRDT-based offline-first synchronization.
 class CrdtSyncEndpoint extends Endpoint {
   @override
@@ -14,7 +16,7 @@ class CrdtSyncEndpoint extends Endpoint {
     required Stream<CrdtSyncStreamEvent> changes,
     bool once = false,
   }) async* {
-    yield* CrdtSync.instance.sync(
+    yield* session.crdt.sync(
       session,
       userId: UuidValue.withValidation(session.authenticated!.userIdentifier),
       inbound: changes,

test/serverpod_offline_sync_test_server/lib/server.dart

@@ -15,12 +15,7 @@ Future<void> run(List<String> args) async {
     Protocol(),
     Endpoints(),
     authenticationHandler: demoAuthenticationHandler,
-    // TODO: Remove the duplicate [syncTables] argument from here and the
+    databaseInterceptor: crdtDatabaseInterceptor,
-    // [initializeCrdtSync] call below.
-    databaseInterceptor: (_, inner) => CrdtDatabase(
-      inner,
-      syncTables: demoSyncTables,
-    ),
   )..initializeCrdtSync(syncTables: demoSyncTables);
 
   // Start the server.

test/serverpod_offline_sync_test_server/test/integration/server/crdt_database_interceptor_test.dart

@@ -0,0 +1,70 @@
+import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
+import 'package:serverpod_offline_sync_test_server/src/generated/protocol.dart'
+    as server;
+import 'package:test/test.dart';
+
+import '../test_tools/serverpod_test_tools.dart';
+
+void main() {
+  final syncTables = [server.Person.t, server.Address.t];
+
+  withServerpod(
+    'Given a Serverpod with CRDT sync initialized,',
+    (sessionBuilder, _) {
+      final session = sessionBuilder.build();
+      session.serverpod.initializeCrdtSync(syncTables: syncTables);
+
+      test(
+        'when the database interceptor wraps a session database, '
+        'then it returns a CRDT-aware database.',
+        () {
+          final wrapped = crdtDatabaseInterceptor(session, session.db);
+          expect(wrapped, isA<CrdtDatabase>());
+        },
+      );
+
+      test(
+        'when the database interceptor wraps an already-wrapped database, '
+        'then it returns the same instance.',
+        () {
+          final wrapped = crdtDatabaseInterceptor(session, session.db);
+          final reWrapped = crdtDatabaseInterceptor(session, wrapped);
+          expect(identical(reWrapped, wrapped), isTrue);
+        },
+      );
+
+      test(
+        'when resolving the CRDT sync for the pod, '
+        'then the configured instance is returned.',
+        () {
+          expect(session.crdt, isNotNull);
+        },
+      );
+    },
+  );
+
+  withServerpod(
+    'Given a Serverpod without CRDT sync initialized,',
+    (sessionBuilder, _) {
+      final session = sessionBuilder.build();
+
+      test(
+        'when the database interceptor wraps a session database, '
+        'then it returns the original database unchanged.',
+        () {
+          final inner = session.db;
+          final result = crdtDatabaseInterceptor(session, inner);
+          expect(identical(result, inner), isTrue);
+        },
+      );
+
+      test(
+        'when resolving the CRDT sync for the pod, '
+        'then a StateError is thrown.',
+        () {
+          expect(() => session.crdt, throwsStateError);
+        },
+      );
+    },
+  );
+}

test/serverpod_offline_sync_test_server/test/integration/sync/sync_flow_test.dart

@@ -165,7 +165,7 @@ void main() {
                   clientToServer.add(event);
                 });
 
-            final serverSubscription = CrdtSync.instance
+            final serverSubscription = rawServerSession.crdt
                 .sync(
                   serverSession,
                   userId: testCrdtUserId,