feat: Write a row in its own space, and a new one where the host says

A client with a persistent user wrote every change in its personal
space: an update or delete of a row in a shared space it is a member of
found no row and threw, and a new row could reach a shared space only
through transactionForUser, which a host calls at each write itself.

A write without a transaction now acts in the space it touches, through
transactionForUser, which checks membership and role as before: an
update or delete in the space its rows are stored in, a new row in the
one `newRowSpace` names (null, or no hook, for the personal space). One
call touches one space and is refused, with a StateError, where it
would touch two. A write within a transaction acts in its space, as
before; the server, without a persistent user, is unchanged.

Tested on the client: a writing member updates and deletes a shared
row, by row and by condition; a reading member is refused; a call
naming rows of two spaces is refused; new rows go where the hook says,
refused for a space only read or not a member's; a transaction of the
caller's own still decides. The integration suite passes (994 tests).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Tizian 「ティツィアーン」 <raisondetredev@pm.me>

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

@@ -85,6 +85,19 @@ class OfflineSyncDatabase implements Database {
 
   final CrdtMutationRecorder _recorder;
 
+  /// Which space a new row goes in, for a client with a persistent user
+  /// writing without a transaction: a space the user is a member of, or
+  /// null for their personal space. Without it, new rows go in the
+  /// personal space.
+  ///
+  /// A write without a transaction acts in the space it touches: rows
+  /// updated or deleted in their own space, new rows in the one this
+  /// names, each through [transactionForUser], which checks that the user
+  /// may write there. One call touches one space, and is refused
+  /// ([StateError]) where it would touch two. A write within a transaction
+  /// acts in that transaction's space, as before.
+  FutureOr<UuidValue?> Function(TableRow row)? newRowSpace;
+
   late final _sync = OfflineSyncEngine(
     syncTables: _syncTables,
     serializationManager: serializationManager,
@@ -343,6 +356,21 @@ class OfflineSyncDatabase implements Database {
         noReturn: noReturn,
       );
     }
+    if (_routesWrite(transaction)) {
+      final space = await _newRowsSpace(rows);
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => insert<T>(
+            rows,
+            transaction: tx,
+            ignoreConflicts: ignoreConflicts,
+            noReturn: noReturn,
+          ),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {
@@ -443,6 +471,31 @@ class OfflineSyncDatabase implements Database {
         noReturn: noReturn,
       );
     }
+    if (_routesWrite(transaction)) {
+      final table = rows.first.table;
+      final ids = {
+        for (final row in rows)
+          if (row.id case final Object id) id,
+      };
+      final stored = ids.isEmpty
+          ? null
+          : await _storedRowsSpace<T>(table.id.inSet(ids.castToIdType().toSet()));
+      final space = stored ?? await _newRowsSpace(rows);
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => upsert<T>(
+            rows,
+            conflictColumns: conflictColumns,
+            updateColumns: updateColumns,
+            updateWhere: updateWhere,
+            transaction: tx,
+            noReturn: noReturn,
+          ),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {
@@ -663,6 +716,23 @@ class OfflineSyncDatabase implements Database {
         noReturn: noReturn,
       );
     }
+    if (_routesWrite(transaction)) {
+      final space = await _storedRowsSpace<T>(
+        rows.first.table.id.inSet(rows.map((row) => row.id).castToIdType().toSet()),
+      );
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => update<T>(
+            rows,
+            columns: columns,
+            transaction: tx,
+            noReturn: noReturn,
+          ),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {
@@ -703,6 +773,16 @@ class OfflineSyncDatabase implements Database {
         transaction: transaction,
       );
     }
+    if (_routesWrite(transaction)) {
+      final space = await _storedRowsSpace<T>(row.table.id.equals(row.id));
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => updateRow<T>(row, columns: columns, transaction: tx),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {
@@ -808,6 +888,26 @@ class OfflineSyncDatabase implements Database {
     }
 
     _assertNoSpaceIdColumnValues<T>(columnValues);
+    if (_routesWrite(transaction)) {
+      final space = await _storedRowsSpace<T>(where);
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => updateWhere<T>(
+            columnValues: columnValues,
+            where: where,
+            limit: limit,
+            offset: offset,
+            orderBy: orderBy,
+            orderByList: orderByList,
+            orderDescending: orderDescending,
+            transaction: tx,
+            noReturn: noReturn,
+          ),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {
@@ -917,6 +1017,23 @@ class OfflineSyncDatabase implements Database {
       );
     }
 
+    if (_routesWrite(transaction)) {
+      final space = await _storedRowsSpace<T>(where);
+      if (space != null) {
+        return transactionForUser(
+          _recorder.persistentUserId!,
+          (tx) => deleteWhere<T>(
+            where: where,
+            orderBy: orderBy,
+            orderByList: orderByList,
+            orderDescending: orderDescending,
+            transaction: tx,
+            noReturn: noReturn,
+          ),
+          spaceId: space,
+        );
+      }
+    }
     return _runTrackedWrite(
       transaction,
       (tx) async {

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

@@ -80,6 +80,63 @@ extension _OfflineSyncDatabaseSpace on OfflineSyncDatabase {
     }
   }
 
+  /// Whether a write chooses its space itself ([OfflineSyncDatabase.newRowSpace]):
+  /// one without a transaction, on a client with a persistent user.
+  bool _routesWrite(Transaction? transaction) =>
+      transaction == null && _recorder.persistentUserId != null;
+
+  /// The space other than the personal one that new [rows] go in, as
+  /// [OfflineSyncDatabase.newRowSpace] says; null for the personal one.
+  /// Throws a [StateError] where the rows go in two spaces.
+  Future<UuidValue?> _newRowsSpace<T extends TableRow>(List<T> rows) async {
+    final route = newRowSpace;
+    if (route == null) return null;
+    final personal = _recorder.persistentUserId!;
+    final spaces = <UuidValue>{
+      for (final row in rows)
+        if (_readSpaceId(row) == null) await route(row) ?? personal,
+    };
+    if (spaces.length > 1) {
+      throw StateError(
+        'Cannot write new ${rows.first.table.tableName} rows into the spaces '
+        '$spaces in one call.',
+      );
+    }
+    final space = spaces.singleOrNull;
+    return space == personal ? null : space;
+  }
+
+  /// The space other than the personal one that the visible rows matching
+  /// [where] are stored in; null where they are personal, or none match.
+  /// Throws a [StateError] where they are stored in two spaces.
+  Future<UuidValue?> _storedRowsSpace<T extends TableRow>(Expression where) async {
+    final rows = await _delegate.find<T>(
+      where: await _whereVisibleWithTombstone<T>(
+        where,
+        null,
+        null,
+        membershipWide: true,
+      ),
+    );
+    final spaceIds = {
+      for (final row in rows)
+        if (_readSpaceId(row) case final int spaceId) spaceId,
+    };
+    if (spaceIds.isEmpty) return null;
+    if (spaceIds.length > 1) {
+      throw StateError(
+        'Cannot write ${rows.first.table.tableName} rows of several spaces in '
+        'one call.',
+      );
+    }
+    final space = await OfflineSyncSpace.db.findById(
+      _delegate.session,
+      spaceIds.single,
+    );
+    final uuid = space?.uuidSpaceId;
+    return uuid == null || uuid == _recorder.persistentUserId ? null : uuid;
+  }
+
   OfflineSyncSpace _requireEffectiveSpace(Transaction transaction) {
     return _recorder.spaceForQueries(transaction) ??
         (throw StateError(

test/serverpod_offline_sync_test_server/test/integration/crud/space_routing_test.dart

@@ -0,0 +1,217 @@
+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';
+
+import '../test_tools/client_session.dart';
+
+void main() {
+  initTestClientSession(withPersistentUser: true);
+
+  group(
+    'Given a persistent user who is a member of a shared space holding a row, '
+    'and a row of their own,',
+    () {
+      late UuidValue sharedSpaceUuid;
+      late OfflineSyncSpace sharedSpace;
+      late Person shared;
+      late Person own;
+
+      /// The space each person row is stored in, by name, read past the
+      /// engine.
+      Future<Map<String, int?>> storedSpaces() async => {
+        for (final person in await Person.db.find(testSession))
+          person.name: person.spaceId,
+      };
+
+      Future<void> joinAs(OfflineSyncSpaceRole role) async {
+        await OfflineSyncSpaceMember.db.insertRow(
+          session,
+          OfflineSyncSpaceMember(
+            spaceId: sharedSpace.id!,
+            userUuid: testCrdtUserId,
+            role: role,
+          ),
+        );
+      }
+
+      setUp(() async {
+        sharedSpaceUuid = const Uuid().v7obj();
+        shared = await session.db.transactionForUser(
+          sharedSpaceUuid,
+          (tx) => Person.db.insertRow(
+            session,
+            Person(name: 'shared'),
+            transaction: tx,
+          ),
+        );
+        sharedSpace = (await OfflineSyncSpace.db.findFirstRow(
+          testSession,
+          where: (t) => t.uuidSpaceId.equals(sharedSpaceUuid),
+        ))!;
+        own = await Person.db.insertRow(session, Person(name: 'own'));
+      });
+
+      test(
+        'when a writing member updates and deletes the shared row without a '
+        'transaction, then it changes in its own space.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+
+          final updated = await Person.db.updateRow(
+            session,
+            shared.copyWith(name: 'shared, changed'),
+          );
+          expect(updated.name, 'shared, changed');
+          expect((await storedSpaces())['shared, changed'], sharedSpace.id);
+
+          await Person.db.deleteRow(session, updated);
+          expect(await Person.db.findById(session, shared.id!), isNull);
+          expect(await Person.db.findById(session, own.id!), isNotNull);
+        },
+      );
+
+      test(
+        'when a writing member updates the shared rows by a condition, then '
+        'they change in their own space.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+
+          await Person.db.updateWhere(
+            session,
+            columnValues: (t) => [t.name('renamed')],
+            where: (t) => t.name.equals('shared'),
+          );
+          await Person.db.deleteWhere(
+            session,
+            where: (t) => t.name.equals('renamed'),
+          );
+
+          expect(
+            [for (final p in await Person.db.find(session)) p.name],
+            ['own'],
+          );
+        },
+      );
+
+      test(
+        'when a reading member updates or deletes the shared row, then it is '
+        'refused and stays as it was.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readOnly);
+
+          await expectLater(
+            Person.db.updateRow(session, shared.copyWith(name: 'changed')),
+            throwsA(isA<OfflineSyncSpaceRoleException>()),
+          );
+          await expectLater(
+            Person.db.deleteRow(session, shared),
+            throwsA(isA<OfflineSyncSpaceRoleException>()),
+          );
+          expect((await Person.db.findById(session, shared.id!))?.name, 'shared');
+        },
+      );
+
+      test(
+        'when one call names rows of two spaces, then it is refused and '
+        'nothing changes.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+
+          await expectLater(
+            Person.db.update(session, [
+              shared.copyWith(name: 'a'),
+              own.copyWith(name: 'b'),
+            ]),
+            throwsA(isA<StateError>()),
+          );
+          expect(
+            {for (final p in await Person.db.find(session)) p.name},
+            {'shared', 'own'},
+          );
+        },
+      );
+
+      test(
+        'when the host routes a new row to the shared space, then it is '
+        'stored there; one it routes nowhere stays personal.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+          session.db.newRowSpace = (row) =>
+              (row as Person).name.startsWith('to shared') ? sharedSpaceUuid : null;
+
+          await Person.db.insertRow(session, Person(name: 'to shared'));
+          await Person.db.insert(session, [Person(name: 'to shared, too')]);
+          await Person.db.insertRow(session, Person(name: 'mine'));
+
+          final spaces = await storedSpaces();
+          expect(spaces['to shared'], sharedSpace.id);
+          expect(spaces['to shared, too'], sharedSpace.id);
+          expect(spaces['mine'], spaces['own']);
+        },
+      );
+
+      test(
+        'when the host routes new rows of one call to two spaces, then it is '
+        'refused.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+          session.db.newRowSpace = (row) =>
+              (row as Person).name == 'there' ? sharedSpaceUuid : null;
+
+          await expectLater(
+            Person.db.insert(session, [Person(name: 'there'), Person(name: 'here')]),
+            throwsA(isA<StateError>()),
+          );
+          expect(
+            {for (final p in await Person.db.find(session)) p.name},
+            {'shared', 'own'},
+          );
+        },
+      );
+
+      test(
+        'when the host routes a new row to a space the person may only read, '
+        'or is not a member of, then it is refused.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readOnly);
+          final elsewhere = const Uuid().v7obj();
+          session.db.newRowSpace = (row) =>
+              (row as Person).name == 'read only' ? sharedSpaceUuid : elsewhere;
+
+          await expectLater(
+            Person.db.insertRow(session, Person(name: 'read only')),
+            throwsA(isA<OfflineSyncSpaceRoleException>()),
+          );
+          await expectLater(
+            Person.db.insertRow(session, Person(name: 'nowhere')),
+            throwsA(isA<OfflineSyncSpaceMembershipException>()),
+          );
+          expect(
+            {for (final p in await Person.db.find(session)) p.name},
+            {'shared', 'own'},
+          );
+        },
+      );
+
+      test(
+        'when a caller writes within a transaction of its own, then the '
+        'transaction decides the space, as before.',
+        () async {
+          await joinAs(OfflineSyncSpaceRole.readWrite);
+          session.db.newRowSpace = (_) => sharedSpaceUuid;
+
+          await session.db.transaction((tx) async {
+            await Person.db.insertRow(
+              session,
+              Person(name: 'in a transaction'),
+              transaction: tx,
+            );
+          });
+
+          final spaces = await storedSpaces();
+          expect(spaces['in a transaction'], spaces['own']);
+        },
+      );
+    },
+  );
+}