halkyon-days/serverpod-offline-sync
Watch
1
0
Fork
You've already forked serverpod-offline-sync
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: Require scopeId on domain tables to ensure user isolation (#73)

Browse source
This commit is contained in:
Marcelo Mendonça Soares 2026-06-18 01:05:24 -03:00 • committed by GitHub
commit 84cfbadc45
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 4258 additions and 770 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/serverpod_offline_sync_client/lib/crdt.dart
View file

@ -1,5 +1,5 @@
export 'src/crdt/client_sync.dart';
export 'src/crdt/exceptions.dart';
export 'src/crdt/exceptions.dart' hide PendingOutboundIntegrityViolation;
export 'src/crdt/merge.dart';
export 'src/crdt/sync.dart';
export 'src/database/database.dart';
@ -7,4 +7,4 @@ export 'src/database/recorder.dart';
export 'src/database/schema.dart';
export 'src/database/session.dart' show CrdtDatabaseAccess, CrdtDatabaseSession;
export 'src/managers/hlc.dart';
export 'src/managers/user.dart';
export 'src/managers/scope.dart';

6
packages/serverpod_offline_sync_client/lib/migrations/analysis_options.yaml generated Normal file
View file

@ -0,0 +1,6 @@
analyzer:
errors:
camel_case_types: ignore
formatter:
trailing_commas: preserve

56
packages/serverpod_offline_sync_client/lib/src/crdt/exceptions.dart
View file

@ -1,3 +1,5 @@
import 'package:meta/meta.dart';
import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart';
import 'package:serverpod_serialization/serverpod_serialization.dart';
import '../protocol/protocol.dart';
@ -82,6 +84,60 @@ final class SyncTablesHashMismatchException extends CrdtSyncException {
'schema version before syncing.';
}
/// Thrown when sync observes terminal CRDT metadata/domain-row inconsistency.
final class CrdtSyncIntegrityViolationException extends CrdtSyncException {
/// Creates a [CrdtSyncIntegrityViolationException].
const CrdtSyncIntegrityViolationException(this.violation);
/// The integrity violation that caused sync to stop.
final CrdtSyncIntegrityViolation violation;
@override
String toString() {
final persisted = violation.id == null
? ''
: ' Persisted violation id: ${violation.id}.';
final row = '${violation.domainTableName}.${violation.uuidRowId}';
return switch (violation.type) {
CrdtSyncViolationType.ownershipCollision =>
'CrdtSyncIntegrityViolationException: '
'${violation.type.name}/${violation.operation.name} for $row '
'belongs to scope ${violation.ownerScopeUuid}, but sync attempted '
'scope ${violation.incomingScopeUuid}.$persisted',
CrdtSyncViolationType.missingDomainRow =>
'CrdtSyncIntegrityViolationException: '
'${violation.type.name}/${violation.operation.name} references '
'missing domain row $row for scope '
'${violation.incomingScopeUuid}.$persisted',
};
}
}
@internal
final class PendingOutboundIntegrityViolation implements Exception {
const PendingOutboundIntegrityViolation({
required this.crdtDataRowId,
required this.type,
required this.operation,
required this.tableName,
required this.rowId,
required this.ownerScopeId,
required this.incomingScopeUuid,
required this.uuidNodeId,
this.hlc,
});
final int? crdtDataRowId;
final CrdtSyncViolationType type;
final CrdtSyncViolationOperation operation;
final String tableName;
final UuidValue rowId;
final int? ownerScopeId;
final UuidValue incomingScopeUuid;
final UuidValue uuidNodeId;
final Hlc? hlc;
}
extension on Type {
/// The name of the class without the leading underscore and without the 'Impl' suffix.
String get className {

54
packages/serverpod_offline_sync_client/lib/src/crdt/integrity_violation.dart Normal file
View file

@ -0,0 +1,54 @@
import 'package:serverpod_database/serverpod_database.dart';
import '../protocol/protocol.dart';
/// Persists or updates the durable row for an integrity violation.
Future<CrdtSyncIntegrityViolation> recordCrdtSyncIntegrityViolation(
DatabaseSession session, {
required CrdtSyncIntegrityViolation violation,
Transaction? transaction,
}) async {
final existing = await CrdtSyncIntegrityViolation.db.findFirstRow(
session,
where: (t) =>
t.type.equals(violation.type) &
t.operation.equals(violation.operation) &
t.domainTableName.equals(violation.domainTableName) &
t.uuidRowId.equals(violation.uuidRowId) &
t.ownerScopeUuid.equals(violation.ownerScopeUuid) &
t.incomingScopeUuid.equals(violation.incomingScopeUuid),
transaction: transaction,
);
final now = DateTime.now().toUtc();
final newViolation = existing != null
? existing.copyWith(
crdtDataRowId: violation.crdtDataRowId ?? existing.crdtDataRowId,
uuidNodeId: violation.uuidNodeId ?? existing.uuidNodeId,
hlcDatetime: violation.hlcDatetime,
hlcCounter: violation.hlcCounter,
lastSeenAt: now,
occurrences: existing.occurrences + 1,
)
: violation.copyWith(
firstSeenAt: now,
lastSeenAt: now,
occurrences: 1,
);
final persisted = await CrdtSyncIntegrityViolation.db.upsertRow(
session,
newViolation,
conflictColumns: (t) => [
t.type,
t.operation,
t.domainTableName,
t.uuidRowId,
t.ownerScopeUuid,
t.incomingScopeUuid,
],
transaction: transaction,
);
return persisted!;
}

301
packages/serverpod_offline_sync_client/lib/src/crdt/sync.dart
View file

@ -5,16 +5,23 @@ import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart
import 'package:uuid/uuid.dart';
import '../database/database.dart';
import '../managers/user.dart';
import '../managers/scope.dart';
import '../protocol/protocol.dart';
import '../utils/case_when.dart' show Case;
import 'exceptions.dart';
import 'extensions.dart';
import 'integrity_violation.dart';
import 'merge.dart';
/// Callback function for when a merge is successful.
typedef CrdtSyncOnMergeSuccess = FutureOr<void> Function(Hlc syncedHlc);
/// A tuple representing the ownership of a domain row.
typedef DomainRowOwner = ({bool exists, int? scopeId});
/// A cache of domain row owners by table name and row id.
typedef DomainRowOwnerCache = Map<(String, UuidValue), DomainRowOwner>;
/// The shared CRDT synchronization logic used by both client and server nodes.
class CrdtSync {
/// Creates a new [CrdtSync] instance.
@ -142,10 +149,18 @@ class CrdtSync {
required UuidValue peerNodeId,
required List<Hlc> nodeCheckpoints,
}) async* {
final crdtUser = await CrdtUserManager(session).getOrCreate(userId);
yield* _streamInserts(session, crdtUser, nodeCheckpoints);
yield* _streamUpdates(session, crdtUser, nodeCheckpoints);
yield* _streamDeletes(session, crdtUser, nodeCheckpoints);
final crdtUser = await CrdtScopeManager(session).getOrCreate(userId);
try {
await for (final change in _streamPendingChanges(
session,
crdtUser,
nodeCheckpoints,
)) {
yield change;
}
} on PendingOutboundIntegrityViolation catch (violation) {
await _recordAndThrowIntegrityViolation(session, violation);
}
}
/// Creates the [CrdtSyncSinceHlc] checkpoint for the post-connect handshake.
@ -157,12 +172,12 @@ class CrdtSync {
required UuidValue userId,
required UuidValue peerNodeId,
}) async {
final crdtUser = await CrdtUserManager(session).getOrCreate(userId);
final crdtUser = await CrdtScopeManager(session).getOrCreate(userId);
final nodes = await CrdtNode.db.find(
session,
where: (t) =>
t.userId.equals(crdtUser.id) &
t.scopeId.equals(crdtUser.id) &
t.uuidNodeId.notEquals(crdtUser.currentNode!.uuidNodeId),
);
@ -194,7 +209,7 @@ class CrdtSync {
if (mergeSet.isEmpty) return null;
final maxSyncedHlc = mergeSet.maxHlc;
final crdtDb = await _openCrdtDatabase(session);
await crdtDb.mergeChanges(mergeSet, userId: userId);
await crdtDb.mergeChanges(mergeSet, scopeId: userId);
if (maxSyncedHlc != null) {
await crdtDb.recordSyncCheckpoint(otherNodeId, maxSyncedHlc, userId: userId);
}
@ -255,7 +270,7 @@ class CrdtSync {
var sessionCompleted = false;
try {
final crdtUser = await CrdtUserManager(session).getOrCreate(userId);
final crdtUser = await CrdtScopeManager(session).getOrCreate(userId);
final localNodeId = crdtUser.currentNode!.uuidNodeId;
yield CrdtSyncConnect(
@ -363,10 +378,24 @@ class CrdtSync {
return crdtDb;
}
Stream<CrdtMergeChange> _streamPendingChanges(
DatabaseSession session,
CrdtScope crdtUser,
List<Hlc> nodeCheckpoints,
) async* {
// Domain ownership is immutable while a collection runs, so read each
// row's owner at most once across all three streams.
final ownerCache = DomainRowOwnerCache();
yield* _streamInserts(session, crdtUser, nodeCheckpoints, ownerCache);
yield* _streamUpdates(session, crdtUser, nodeCheckpoints, ownerCache);
yield* _streamDeletes(session, crdtUser, nodeCheckpoints, ownerCache);
}
Stream<CrdtMergeInsert> _streamInserts(
DatabaseSession session,
CrdtUser crdtUser,
CrdtScope crdtUser,
List<Hlc> nodeCheckpoints,
DomainRowOwnerCache ownerCache,
) async* {
final rows = await CrdtDataRow.db.find(
session,
@ -397,8 +426,35 @@ class CrdtSync {
table,
dartName,
foreignKeyAttemptFieldsByRowId[row.id!],
crdtUser.id!,
ownerCache,
);
if (domainRow == null) continue;
if (!domainRow.exists) {
_throwPendingIntegrityViolation(
crdtDataRowId: row.id,
type: CrdtSyncViolationType.missingDomainRow,
operation: CrdtSyncViolationOperation.outboundInsert,
tableName: tableName,
rowId: row.uuidRowId,
ownerScopeId: null,
incomingScopeUuid: crdtUser.uuidScopeId,
uuidNodeId: row.node!.uuidNodeId,
hlc: row.hlc,
);
}
if (domainRow.ownerScopeId != crdtUser.id) {
_throwPendingIntegrityViolation(
crdtDataRowId: row.id,
type: CrdtSyncViolationType.ownershipCollision,
operation: CrdtSyncViolationOperation.outboundInsert,
tableName: tableName,
rowId: row.uuidRowId,
ownerScopeId: domainRow.ownerScopeId,
incomingScopeUuid: crdtUser.uuidScopeId,
uuidNodeId: row.node!.uuidNodeId,
hlc: row.hlc,
);
}
yield CrdtMergeInsert(
hlcDatetime: row.hlcDatetime,
@ -406,15 +462,16 @@ class CrdtSync {
tableName: tableName,
uuidRowId: row.uuidRowId,
uuidNodeId: row.node!.uuidNodeId,
data: domainRow,
data: domainRow.row,
);
}
}
Stream<CrdtMergeUpdate> _streamUpdates(
DatabaseSession session,
CrdtUser crdtUser,
CrdtScope crdtUser,
List<Hlc> nodeCheckpoints,
DomainRowOwnerCache ownerCache,
) async* {
final fields = await CrdtDataField.db.find(
session,
@ -437,13 +494,41 @@ class CrdtSync {
}
final columnName = field.column!.name;
final decodedValue = await _fetchColumnValue(
final columnValue = await _fetchOwnedColumnValue(
session,
tableName,
field.row!.uuidRowId,
columnName,
field.foreignKey,
crdtUser.id!,
ownerCache,
);
if (!columnValue.exists) {
_throwPendingIntegrityViolation(
crdtDataRowId: field.row!.id,
type: CrdtSyncViolationType.missingDomainRow,
operation: CrdtSyncViolationOperation.outboundUpdate,
tableName: tableName,
rowId: field.row!.uuidRowId,
ownerScopeId: null,
incomingScopeUuid: crdtUser.uuidScopeId,
uuidNodeId: field.node!.uuidNodeId,
hlc: field.hlc,
);
}
if (columnValue.ownerScopeId != crdtUser.id) {
_throwPendingIntegrityViolation(
crdtDataRowId: field.row!.id,
type: CrdtSyncViolationType.ownershipCollision,
operation: CrdtSyncViolationOperation.outboundUpdate,
tableName: tableName,
rowId: field.row!.uuidRowId,
ownerScopeId: columnValue.ownerScopeId,
incomingScopeUuid: crdtUser.uuidScopeId,
uuidNodeId: field.node!.uuidNodeId,
hlc: field.hlc,
);
}
yield CrdtMergeUpdate(
hlcDatetime: field.hlcDatetime,
@ -452,15 +537,16 @@ class CrdtSync {
uuidRowId: field.row!.uuidRowId,
uuidNodeId: field.node!.uuidNodeId,
columnName: columnName,
value: decodedValue,
value: columnValue.value,
);
}
}
Stream<CrdtMergeDelete> _streamDeletes(
DatabaseSession session,
CrdtUser crdtUser,
CrdtScope crdtUser,
List<Hlc> nodeCheckpoints,
DomainRowOwnerCache ownerCache,
) async* {
final tombstones = await CrdtDataDeleted.db.find(
session,
@ -476,6 +562,25 @@ class CrdtSync {
final tableName = tombstone.row!.tbl!.name;
if (!_syncTablesByName.containsKey(tableName)) continue;
final owner = await _readDomainRowOwner(
session,
tableName,
tombstone.row!.uuidRowId,
ownerCache,
);
if (owner.exists && owner.scopeId != crdtUser.id) {
_throwPendingIntegrityViolation(
crdtDataRowId: tombstone.row!.id,
type: CrdtSyncViolationType.ownershipCollision,
operation: CrdtSyncViolationOperation.outboundDelete,
tableName: tableName,
rowId: tombstone.row!.uuidRowId,
ownerScopeId: owner.scopeId,
incomingScopeUuid: crdtUser.uuidScopeId,
uuidNodeId: tombstone.node!.uuidNodeId,
hlc: tombstone.hlc,
);
}
yield CrdtMergeDelete(
hlcDatetime: tombstone.hlcDatetime,
@ -491,10 +596,10 @@ class CrdtSync {
Expression _rowHlcAfterFilter(
CrdtDataRowTable t,
CrdtUser crdtUser,
CrdtScope crdtUser,
List<Hlc> nodeCheckpoints,
) =>
t.userId.equals(crdtUser.id) &
t.scopeId.equals(crdtUser.id) &
t.node.afterAnyCheckpointFilter(
t.node.uuidNodeId,
t.hlcDatetime,
@ -504,10 +609,10 @@ class CrdtSync {
Expression _fieldHlcAfterFilter(
CrdtDataFieldTable t,
CrdtUser crdtUser,
CrdtScope crdtUser,
List<Hlc> nodeCheckpoints,
) =>
t.row.userId.equals(crdtUser.id) &
t.row.scopeId.equals(crdtUser.id) &
t.node.afterAnyCheckpointFilter(
t.node.uuidNodeId,
t.hlcDatetime,
@ -517,10 +622,10 @@ class CrdtSync {
Expression _tombstoneHlcAfterFilter(
CrdtDataDeletedTable t,
CrdtUser crdtUser,
CrdtScope crdtUser,
List<Hlc> nodeCheckpoints,
) =>
t.row.userId.equals(crdtUser.id) &
t.row.scopeId.equals(crdtUser.id) &
t.node.afterAnyCheckpointFilter(
t.node.uuidNodeId,
t.hlcDatetime,
@ -534,33 +639,45 @@ class CrdtSync {
/// with an active override back to [CrdtDataForeignKey.attemptedValue] before
/// deserializing. This is the inverse of inbound FK materialization: the wire
/// payload carries attempted facts, not locally projected visible values.
Future<dynamic> _fetchDomainRow(
Future<({bool exists, int? ownerScopeId, dynamic row})> _fetchDomainRow(
DatabaseSession session,
String tableName,
UuidValue rowId,
Table table,
String dartName,
List<CrdtDataField>? foreignKeyAttemptFields,
int scopeId,
DomainRowOwnerCache ownerCache,
) async {
final cols = table.columns
.map((column) => '"${_escapeIdentifier(column.columnName)}"')
.join(', ');
final encodedRowId = ValueEncoder.instance.convert(rowId);
final encodedScopeId = ValueEncoder.instance.convert(scopeId);
final escapedTableName = _escapeIdentifier(tableName);
final result = await session.db.unsafeQuery(
'SELECT $cols FROM "${_escapeIdentifier(tableName)}" '
'WHERE "id" = $encodedRowId',
'SELECT $cols FROM "$escapedTableName" '
'WHERE "id" = $encodedRowId AND "scopeId" = $encodedScopeId '
'LIMIT 1',
);
if (result.isEmpty) return null;
if (result.isEmpty) {
final owner = await _readDomainRowOwner(session, tableName, rowId, ownerCache);
return (exists: owner.exists, ownerScopeId: owner.scopeId, row: null);
}
ownerCache[(tableName, rowId)] = (exists: true, scopeId: scopeId);
final columnMap = result.first.toColumnMap()
// Domain columns hold visible/materialized FK values; restore attempted
// values for override columns before building the outbound merge payload.
..applyProjectedForeignKeyAttempts(foreignKeyAttemptFields);
..applyProjectedForeignKeyAttempts(foreignKeyAttemptFields)
// scopeId is local ownership metadata; it is never emitted on the wire.
..remove('scopeId');
return session.db.serializationManager.deserializeByClassName({
final row = session.db.serializationManager.deserializeByClassName({
'className': dartName,
'data': columnMap,
});
return (exists: true, ownerScopeId: scopeId, row: row);
}
/// Resolves a column value for outbound update sync.
@ -569,25 +686,135 @@ class CrdtSync {
/// from [CrdtDataForeignKey.attemptedValue] instead of the materialized
/// domain column value. Non-FK columns and FK columns without an override
/// are read directly from the domain table.
Future<dynamic> _fetchColumnValue(
Future<({bool exists, int? ownerScopeId, dynamic value})> _fetchOwnedColumnValue(
DatabaseSession session,
String tableName,
UuidValue rowId,
String columnName,
CrdtDataForeignKey? projection,
int scopeId,
DomainRowOwnerCache ownerCache,
) async {
if (projection != null && projection.hasOverride) {
return _decodeColumnValue(tableName, columnName, projection.attemptedValue);
final owner = await _readDomainRowOwner(session, tableName, rowId, ownerCache);
if (!owner.exists || owner.scopeId != scopeId) {
return (exists: owner.exists, ownerScopeId: owner.scopeId, value: null);
}
return (
exists: true,
ownerScopeId: scopeId,
value: _decodeColumnValue(tableName, columnName, projection.attemptedValue),
);
}
final encodedValue = ValueEncoder.instance.convert(rowId);
final encodedScopeId = ValueEncoder.instance.convert(scopeId);
final escapedTableName = _escapeIdentifier(tableName);
final result = await session.db.unsafeQuery(
'SELECT "${_escapeIdentifier(columnName)}" '
'FROM "${_escapeIdentifier(tableName)}" '
'WHERE "id" = $encodedValue',
'FROM "$escapedTableName" '
'WHERE "id" = $encodedValue AND "scopeId" = $encodedScopeId '
'LIMIT 1',
);
if (result.isEmpty) return null;
return _decodeColumnValue(tableName, columnName, result.first[0]);
if (result.isNotEmpty) {
ownerCache[(tableName, rowId)] = (exists: true, scopeId: scopeId);
return (
exists: true,
ownerScopeId: scopeId,
value: _decodeColumnValue(tableName, columnName, result.first[0]),
);
}
final owner = await _readDomainRowOwner(session, tableName, rowId, ownerCache);
return (exists: owner.exists, ownerScopeId: owner.scopeId, value: null);
}
Future<DomainRowOwner> _readDomainRowOwner(
DatabaseSession session,
String tableName,
UuidValue rowId,
DomainRowOwnerCache ownerCache,
) async {
final cached = ownerCache[(tableName, rowId)];
if (cached != null) return cached;
final encodedValue = ValueEncoder.instance.convert(rowId);
final escapedTableName = _escapeIdentifier(tableName);
final result = await session.db.unsafeQuery(
'SELECT "scopeId" FROM "$escapedTableName" '
'WHERE "id" = $encodedValue '
'LIMIT 1',
);
final owner = result.isEmpty
? (exists: false, scopeId: null)
: (exists: true, scopeId: result.first[0] as int?);
ownerCache[(tableName, rowId)] = owner;
return owner;
}
Never _throwPendingIntegrityViolation({
required int? crdtDataRowId,
required CrdtSyncViolationType type,
required CrdtSyncViolationOperation operation,
required String tableName,
required UuidValue rowId,
required int? ownerScopeId,
required UuidValue incomingScopeUuid,
required UuidValue uuidNodeId,
Hlc? hlc,
}) {
throw PendingOutboundIntegrityViolation(
crdtDataRowId: crdtDataRowId,
type: type,
operation: operation,
tableName: tableName,
rowId: rowId,
ownerScopeId: ownerScopeId,
incomingScopeUuid: incomingScopeUuid,
uuidNodeId: uuidNodeId,
hlc: hlc,
);
}
Future<Never> _recordAndThrowIntegrityViolation(
DatabaseSession session,
PendingOutboundIntegrityViolation pending,
) async {
final ownerScopeUuid = await _scopeUuidForNormalizedId(
session,
pending.ownerScopeId,
);
final now = DateTime.now().toUtc();
final violation = CrdtSyncIntegrityViolation(
type: pending.type,
domainTableName: pending.tableName,
uuidRowId: pending.rowId,
ownerScopeUuid: ownerScopeUuid,
incomingScopeUuid: pending.incomingScopeUuid,
operation: pending.operation,
uuidNodeId: pending.uuidNodeId,
crdtDataRowId: pending.crdtDataRowId,
hlcDatetime: pending.hlc?.datetime,
hlcCounter: pending.hlc?.counter,
firstSeenAt: now,
lastSeenAt: now,
occurrences: 1,
);
final persisted = await recordCrdtSyncIntegrityViolation(
session,
violation: violation,
);
throw CrdtSyncIntegrityViolationException(persisted);
}
Future<UuidValue?> _scopeUuidForNormalizedId(
DatabaseSession session,
int? scopeId,
) async {
if (scopeId == null) return null;
final scope = await CrdtScope.db.findById(session, scopeId);
return scope?.uuidScopeId;
}
/// Loads FK projection metadata for outbound insert sync.
@ -661,8 +888,10 @@ class CrdtSync {
final definition = tableDefinitionsByName[table.tableName];
final columns = [
for (final column in definition?.columns ?? const <ColumnDefinition>[])
column.name,
if (definition == null) ...table.columns.map((column) => column.columnName),
if (column.name != 'scopeId') column.name,
if (definition == null)
for (final column in table.columns)
if (column.columnName != 'scopeId') column.columnName,
]..sort();
final foreignKeys = _canonicalForeignKeys(definition);
final uniqueIndexes = _canonicalUniqueIndexes(definition);

140
packages/serverpod_offline_sync_client/lib/src/database/database.dart
View file

@ -6,6 +6,8 @@ import 'package:serverpod_database/serverpod_database.dart';
import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart';
import 'package:uuid/uuid.dart';
import '../crdt/exceptions.dart';
import '../crdt/integrity_violation.dart';
import '../crdt/merge.dart';
import '../crdt/sync.dart';
import '../protocol/protocol.dart';
@ -13,8 +15,10 @@ import 'recorder.dart';
import 'session.dart';
import 'tombstone.dart';
/// Map of transaction hashes to the user ID they are associated with.
final userForTransaction = <Transaction, CrdtUser>{};
part 'scope.dart';
/// Map of transaction hashes to the scope they are associated with.
final scopeForTransaction = <Transaction, CrdtScope>{};
/// Database proxy that runs insert/update/delete ORM operations inside a
/// transaction to record each change in the CRDT tables.
@ -70,7 +74,7 @@ class CrdtDatabase implements Database {
/// Returns the current node identifier for the effective user.
Future<UuidValue> currentNodeId({UuidValue? userId}) async {
final effectiveUserId = await _requireUserId(userId);
final user = await _recorder.getOrCreateUser(effectiveUserId);
final user = await _recorder.getOrCreateScope(effectiveUserId);
return user.currentNode!.uuidNodeId;
}
@ -105,27 +109,35 @@ class CrdtDatabase implements Database {
);
}
/// Merges remote CRDT changes into the local database for the given user.
/// Merges remote CRDT changes into the local database for the given scope.
///
/// When [userId] is omitted, this uses the recorder's persistent user id.
/// The merge locks the current user's CRDT tables and executes atomically
/// When [scopeId] is omitted, this uses the recorder's persistent user id.
/// The merge locks the current scope's CRDT tables and executes atomically
/// inside [transactionForUser].
Future<void> mergeChanges(
CrdtMergeSet mergeSet, {
UuidValue? userId,
UuidValue? scopeId,
}) async {
if (mergeSet.isEmpty) return;
final effectiveUserId =
userId ??
final effectiveScopeId =
scopeId ??
_recorder.persistentUserId ??
(throw StateError(
'A user ID is required when merging changes without a persistent user.',
'A scope ID is required when merging changes without a persistent user.',
));
await transactionForUser<void>(effectiveUserId, (tx) async {
await _recorder.lockCurrentUser(tx);
await _recorder.mergeChanges(mergeSet, tx);
});
try {
await transactionForUser<void>(effectiveScopeId, (tx) async {
await _recorder.lockCurrentUser(tx);
await _recorder.mergeChanges(mergeSet, tx);
});
} on CrdtSyncIntegrityViolationException catch (exception) {
final persistedViolation = await recordCrdtSyncIntegrityViolation(
_delegate.session,
violation: exception.violation,
);
throw CrdtSyncIntegrityViolationException(persistedViolation);
}
}
@override
@ -138,6 +150,24 @@ class CrdtDatabase implements Database {
DatabaseSerializationManager get serializationManager =>
_delegate.serializationManager;
/// Merges [where] with the CRDT visibility predicates scoped to the queried
/// tables and the user associated with [transaction] (or the persistent
/// user). Without a user scope (e.g. server-side admin reads), a row is
/// only hidden when every user tracking it has it hidden.
Expression? _whereVisibleWithTombstone<T extends TableRow>(
Expression? where,
Include? include,
Transaction? transaction,
) {
return mergeWhereWithTombstone<T>(
serializationManager,
where,
include,
tableIdForName: _recorder.tableIdForName,
scopeId: () => _recorder.scopeForQueries(transaction)?.id,
);
}
@override
Future<List<T>> find<T extends TableRow>({
Expression? where,
@ -151,8 +181,8 @@ class CrdtDatabase implements Database {
LockMode? lockMode,
LockBehavior? lockBehavior,
}) async {
return _delegate.find<T>(
where: mergeWhereWithTombstone<T>(serializationManager, where, include),
final result = await _delegate.find<T>(
where: _whereVisibleWithTombstone<T>(where, include, transaction),
limit: limit,
offset: offset,
orderBy: orderBy,
@ -165,6 +195,7 @@ class CrdtDatabase implements Database {
lockMode: lockMode,
lockBehavior: lockBehavior,
);
return _stripScopeIdFromScopedRead(result, include, transaction);
}
@override
@ -177,13 +208,15 @@ class CrdtDatabase implements Database {
}) async {
final table = serializationManager.getTableForType(T);
final where = table?.id.equals(id);
return _delegate.findFirstRow<T>(
where: mergeWhereWithTombstone<T>(serializationManager, where, include),
final result = await _delegate.findFirstRow<T>(
where: _whereVisibleWithTombstone<T>(where, include, transaction),
transaction: transaction,
include: include,
lockMode: lockMode,
lockBehavior: lockBehavior,
);
if (result == null) return null;
return _stripScopeIdFromScopedRead([result], include, transaction).single;
}
@override
@ -198,8 +231,8 @@ class CrdtDatabase implements Database {
LockMode? lockMode,
LockBehavior? lockBehavior,
}) async {
return _delegate.findFirstRow<T>(
where: mergeWhereWithTombstone<T>(serializationManager, where, include),
final result = await _delegate.findFirstRow<T>(
where: _whereVisibleWithTombstone<T>(where, include, transaction),
offset: offset,
orderBy: orderBy,
orderByList: orderByList,
@ -211,6 +244,8 @@ class CrdtDatabase implements Database {
lockMode: lockMode,
lockBehavior: lockBehavior,
);
if (result == null) return null;
return _stripScopeIdFromScopedRead([result], include, transaction).single;
}
@override
@ -219,17 +254,20 @@ class CrdtDatabase implements Database {
Transaction? transaction,
bool ignoreConflicts = false,
}) async {
if (rows.isEmpty) return [];
return DatabaseUtil.runInTransactionOrSavepoint(
_delegate,
transaction,
(tx) async {
final prepared = _prepareRowsForInsert(rows, tx);
final result = await _delegate.insert<T>(
rows,
prepared.rows,
transaction: tx,
ignoreConflicts: ignoreConflicts,
);
await _recorder.afterInsert<T>(result, tx);
_stripStampedRows(result, prepared);
return result;
},
);
@ -244,18 +282,22 @@ class CrdtDatabase implements Database {
_delegate,
transaction,
(tx) async {
final prepared = _prepareRowsForInsert([row], tx);
final databaseRow = prepared.rows.single;
late final T result;
try {
result = await _delegate.insertRow<T>(row, transaction: tx);
result = await _delegate.insertRow<T>(databaseRow, transaction: tx);
} on DatabaseQueryException {
if (!await _recorder.isDeleted(row, tx)) rethrow;
result = await _delegate.updateRow<T>(row, transaction: tx);
result = await _delegate.updateRow<T>(databaseRow, transaction: tx);
await _recorder.afterReinsert([result], tx);
_stripStampedRows([result], prepared);
return result;
}
await _recorder.afterInsert([result], tx);
_stripStampedRows([result], prepared);
return result;
},
);
@ -281,6 +323,15 @@ class CrdtDatabase implements Database {
Expression? updateWhere,
Transaction? transaction,
}) async {
if (T == CrdtSyncIntegrityViolation) {
return _delegate.upsertRow<T>(
row,
conflictColumns: conflictColumns,
updateColumns: updateColumns,
updateWhere: updateWhere,
transaction: transaction,
);
}
// TODO: Implement CRDT upsert.
throw UnimplementedError('CRDT upsert is not implemented.');
}
@ -298,7 +349,12 @@ class CrdtDatabase implements Database {
(tx) async {
final updatedRows = [
for (final row in rows)
await _updateRowWithoutRecording(row, columns: columns, transaction: tx),
await _updateRowWithoutRecording(
row,
stripScopeId: _shouldStripReturnedScopeId(row, tx),
transaction: tx,
columns: columns,
),
];
await _recorder.afterUpdate(updatedRows, columns, tx);
@ -317,10 +373,12 @@ class CrdtDatabase implements Database {
_delegate,
transaction,
(tx) async {
final stripScopeId = _shouldStripReturnedScopeId(row, tx);
final updatedRow = await _updateRowWithoutRecording(
row,
columns: columns,
stripScopeId: stripScopeId,
transaction: tx,
columns: columns,
);
await _recorder.afterUpdate([updatedRow], columns, tx);
@ -332,18 +390,19 @@ class CrdtDatabase implements Database {
Future<T> _updateRowWithoutRecording<T extends TableRow>(
T row, {
required Transaction transaction,
required bool stripScopeId,
List<Column>? columns,
}) async {
final values = row.toJsonForDatabase() as Map<String, dynamic>;
final columnValues = (columns ?? row.table.managedColumns)
.where((c) => c.columnName != 'id')
.where((c) => c.columnName != 'id' && c.columnName != 'scopeId')
.map((c) => ColumnValue(c, values[c.columnName]))
.toList();
final where = row.table.id.equals(row.id);
final updatedRows = await _delegate.updateWhere<T>(
columnValues: columnValues,
where: mergeWhereWithTombstone<T>(serializationManager, where, null)!,
where: _whereVisibleWithTombstone<T>(where, null, transaction)!,
transaction: transaction,
);
@ -353,7 +412,9 @@ class CrdtDatabase implements Database {
throw Exception('Failed to update row, no rows updated.');
}
return updatedRows.single;
final updatedRow = updatedRows.single;
if (stripScopeId) _stripScopeId(updatedRow);
return updatedRow;
}
@override
@ -386,13 +447,17 @@ class CrdtDatabase implements Database {
bool orderDescending = false,
Transaction? transaction,
}) async {
if (_recorder.isCrdtTracked<T>()) {
_assertNoScopeIdColumnValues<T>(columnValues);
}
return DatabaseUtil.runInTransactionOrSavepoint(
_delegate,
transaction,
(tx) async {
final result = await _delegate.updateWhere<T>(
columnValues: columnValues,
where: mergeWhereWithTombstone<T>(serializationManager, where, null)!,
where: _whereVisibleWithTombstone<T>(where, null, tx)!,
limit: limit,
offset: offset,
orderBy: orderBy,
@ -405,6 +470,9 @@ class CrdtDatabase implements Database {
final columns = columnValues.map((e) => e.column).toList();
await _recorder.afterUpdate(result, columns, tx);
if (_recorder.isCrdtTracked<T>()) {
result.forEach(_stripScopeId);
}
return result;
},
);
@ -476,7 +544,7 @@ class CrdtDatabase implements Database {
transaction,
(tx) async {
final rows = await _delegate.find<T>(
where: mergeWhereWithTombstone<T>(serializationManager, where, null),
where: _whereVisibleWithTombstone<T>(where, null, tx),
orderBy: orderBy,
orderByList: orderByList,
// Remove this once the deprecated member is removed.
@ -486,7 +554,7 @@ class CrdtDatabase implements Database {
);
await _recorder.insteadOfDelete<T>(rows, tx);
return rows;
return _stripScopeIdFromScopedRead(rows, null, tx);
},
);
}
@ -499,7 +567,7 @@ class CrdtDatabase implements Database {
Transaction? transaction,
}) async {
return _delegate.count<T>(
where: mergeWhereWithTombstone<T>(serializationManager, where, null),
where: _whereVisibleWithTombstone<T>(where, null, transaction),
limit: limit,
useCache: useCache,
transaction: transaction,
@ -514,7 +582,7 @@ class CrdtDatabase implements Database {
LockBehavior lockBehavior = LockBehavior.wait,
}) async {
return _delegate.lockRows<T>(
where: mergeWhereWithTombstone<T>(serializationManager, where, null)!,
where: _whereVisibleWithTombstone<T>(where, null, transaction)!,
lockMode: lockMode,
transaction: transaction,
lockBehavior: lockBehavior,
@ -543,15 +611,15 @@ class CrdtDatabase implements Database {
TransactionSettings? settings,
}) async {
// Ensure that the user exists with a node before starting the transaction.
final user = await _recorder.getOrCreateUser(userId);
final user = await _recorder.getOrCreateScope(userId);
return transaction<R>(
(tx) async {
try {
userForTransaction[tx] = user;
scopeForTransaction[tx] = user;
return await transactionFunction(tx);
} finally {
userForTransaction.remove(tx);
scopeForTransaction.remove(tx);
}
},
settings: settings,

321
packages/serverpod_offline_sync_client/lib/src/database/merge.dart
View file

@ -2,11 +2,13 @@ part of 'recorder.dart';
typedef _MergeRowKey = (String, UuidValue);
typedef _MergeFieldKey = (String, UuidValue, String);
typedef _DomainOwner = ({bool exists, int? scopeId});
typedef _MergeContext = ({
Map<_MergeRowKey, CrdtDataRow> rows,
Map<_MergeFieldKey, CrdtDataField> fields,
Map<_MergeFieldKey, Hlc> incomingFieldHlcs,
Map<_MergeRowKey, CrdtDataDeleted> tombstones,
Map<_MergeRowKey, _DomainOwner> domainOwners,
});
/// Adds merge-specific behavior to [CrdtMutationRecorder].
@ -16,10 +18,10 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
/// Locks the current user row so merges can serialize with other work.
Future<void> lockCurrentUser(Transaction transaction) async {
final user = _getEffectiveUser(transaction);
final user = _getEffectiveScope(transaction);
// Use a row lock without fetching the record since the merge path only
// needs serialization against concurrent work for the same user.
await CrdtUser.db.lockRows(
await CrdtScope.db.lockRows(
_session,
where: (t) => t.id.equals(user.id),
transaction: transaction,
@ -39,7 +41,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
final shouldProjectForeignKeys = _mergeOperationsMayAffectForeignKeys(
operations,
);
final currentUser = _getEffectiveUser(transaction);
final currentUser = _getEffectiveScope(transaction);
final remoteNodes = await _findOrCreateNodesForMerge(
currentUser.id!,
{for (final change in operations) change.uuidNodeId},
@ -150,7 +152,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
var nodes = await CrdtNode.db.find(
_session,
where: (t) => t.userId.equals(userId) & t.uuidNodeId.inSet(nodeIds),
where: (t) => t.scopeId.equals(userId) & t.uuidNodeId.inSet(nodeIds),
transaction: transaction,
);
final existingNodeIds = nodes.map((node) => node.uuidNodeId).toSet();
@ -162,7 +164,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
[
for (final uuidNodeId in missingNodeIds)
CrdtNode(
userId: userId,
scopeId: userId,
uuidNodeId: uuidNodeId,
),
],
@ -172,7 +174,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
nodes = await CrdtNode.db.find(
_session,
where: (t) => t.userId.equals(userId) & t.uuidNodeId.inSet(nodeIds),
where: (t) => t.scopeId.equals(userId) & t.uuidNodeId.inSet(nodeIds),
transaction: transaction,
);
}
@ -197,6 +199,13 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
(update.tableName, update.uuidRowId, update.columnName): update.hlc,
};
final tombstones = <_MergeRowKey, CrdtDataDeleted>{};
// Owning scope per row, snapshotted once so the per-operation ownership
// checks never re-read the same immutable `scopeId`. Only rows that
// already have a tracker are pre-read here — those are exactly the rows
// whose ownership the update/delete/existing-insert paths verify. New
// rows go through the insert path, which proves ownership at the physical
// write and refreshes this entry afterwards.
final domainOwners = <_MergeRowKey, _DomainOwner>{};
for (final MapEntry(key: tableName, value: rowIds) in rowIdsByTable.entries) {
final loadedRows = await _findCrdtRows(
@ -216,6 +225,22 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
}
}
final trackedRowIds = loadedRows.map((row) => row.uuidRowId).toSet();
if (trackedRowIds.isNotEmpty) {
final ownerValues = await _readDomainColumnValues(
tableName,
trackedRowIds,
['scopeId'],
transaction,
);
for (final rowId in trackedRowIds) {
final values = ownerValues[rowId];
domainOwners[(tableName, rowId)] = values == null
? (exists: false, scopeId: null)
: (exists: true, scopeId: values['scopeId'] as int?);
}
}
final columnNames = columnNamesByTable[tableName];
if (columnNames == null || columnNames.isEmpty || loadedRows.isEmpty) continue;
@ -252,6 +277,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
fields: fields,
incomingFieldHlcs: incomingFieldHlcs,
tombstones: tombstones,
domainOwners: domainOwners,
);
}
@ -273,33 +299,61 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
final data = _sanitizeMergeRowData(insert.tableName, insert.databaseColumns);
if (currentRow == null) {
final resolvedInsert = await _uniqueConflictResolver._resolveForIncomingInsert(
insert,
data,
context,
transaction,
);
final mergingScope = _getEffectiveScope(transaction);
final mergingScopeId = mergingScope.id!;
try {
// Insert-first inside a savepoint: the domain primary key resolves
// creation races atomically, and `_applyMergeInsertForMissingRow`
// raises `_ForeignScopeRowCollision` when the row turns out to be owned
// by another scope. Rolling the savepoint back then leaves no CRDT
// metadata, released unique values, or foreign key attempts behind.
context.rows[rowKey] = await DatabaseUtil.runInTransactionOrSavepoint(
_db,
transaction,
(savepoint) async {
final resolvedInsert = await _uniqueConflictResolver
._resolveForIncomingInsert(insert, data, context, savepoint);
final safeInsert = await _safeIncomingForeignKeyData(
insert.tableName,
insert.uuidRowId,
resolvedInsert.data,
transaction,
);
final safeInsert = await _safeIncomingForeignKeyData(
insert.tableName,
insert.uuidRowId,
resolvedInsert.data,
savepoint,
);
context.rows[rowKey] = await _applyMergeInsertForMissingRow(
insert,
remoteNode,
incomingHlc,
safeInsert.data,
transaction,
);
await _recordForeignKeyInsertAttempts(
insert.tableName,
{insert.uuidRowId},
transaction,
safeInsert.attempts,
);
final insertedRow = await _applyMergeInsertForMissingRow(
insert,
remoteNode,
incomingHlc,
safeInsert.data,
savepoint,
);
await _recordForeignKeyInsertAttempts(
insert.tableName,
{insert.uuidRowId},
savepoint,
safeInsert.attempts,
);
return insertedRow;
},
);
// The row is now owned by the merging scope (freshly inserted, or a
// same-scope recovery), so same-merge field updates resolve ownership
// from the cache instead of re-reading the domain row.
context.domainOwners[rowKey] = (exists: true, scopeId: mergingScopeId);
} on _ForeignScopeRowCollision catch (collision) {
await _throwOwnershipCollision(
operation: CrdtSyncViolationOperation.mergeInsert,
tableName: insert.tableName,
rowId: insert.uuidRowId,
owningScopeId: collision.owningScopeId,
incomingScope: mergingScope,
metadataRowId: null,
node: remoteNode,
hlc: insert.hlc,
transaction: transaction,
);
}
} else {
await _applyMergeInsertForExistingRow(
insert,
@ -322,13 +376,26 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
final rowKey = (update.tableName, update.uuidRowId);
final row = context.rows[rowKey];
if (row == null) return;
final remoteNode = _requireRemoteNode(remoteNodes, update.uuidNodeId);
if (!await _domainRowOwnedByEffectiveScope(
operation: CrdtSyncViolationOperation.mergeUpdate,
tableName: update.tableName,
rowId: update.uuidRowId,
context: context,
transaction: transaction,
metadataRowId: row.id,
node: remoteNode,
hlc: update.hlc,
)) {
return;
}
final shouldApply = await _shouldMergeFieldMetadataIfNewer(
tableName: update.tableName,
rowId: update.uuidRowId,
columnName: update.columnName,
row: row,
remoteNode: _requireRemoteNode(remoteNodes, update.uuidNodeId),
remoteNode: remoteNode,
incomingHlc: update.hlc,
fields: context.fields,
transaction: transaction,
@ -373,6 +440,19 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
final row = context.rows[rowKey];
if (row == null) return;
if (!delete.reason.isSynced) return;
final remoteNode = _requireRemoteNode(remoteNodes, delete.uuidNodeId);
if (!await _domainRowOwnedByEffectiveScope(
operation: CrdtSyncViolationOperation.mergeDelete,
tableName: delete.tableName,
rowId: delete.uuidRowId,
context: context,
transaction: transaction,
metadataRowId: row.id,
node: remoteNode,
hlc: delete.hlc,
)) {
return;
}
final currentTombstone = context.tombstones[rowKey];
final currentClFlag = currentTombstone?.clFlag ?? 1;
@ -382,7 +462,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
context.tombstones[rowKey] = await _upsertMergeTombstone(
row,
_requireRemoteNode(remoteNodes, delete.uuidNodeId),
remoteNode,
delete.hlc,
delete.clFlag,
delete.reason,
@ -412,7 +492,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
final insertedRow = await CrdtDataRow.db.insertRow(
_session,
CrdtDataRow(
userId: _getHlcManager(transaction).normalizedUserId,
scopeId: _getHlcManager(transaction).normalizedScopeId,
tblId: tableId,
uuidRowId: rowId,
nodeId: remoteNode.id!,
@ -555,16 +635,41 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
final tableRow = _requireMergeInsertTableRow(insert);
final patchedRow = _db.serializationManager.patchTableRow(tableRow, data);
final mergingScopeId = _getHlcManager(transaction).normalizedScopeId;
final scopedRow = _withScopeId(patchedRow, mergingScopeId);
try {
// Insert first: the domain primary key resolves creation races
// atomically. The inner savepoint keeps a violation from poisoning the
// surrounding transaction so the recovery update below can still run.
await DatabaseUtil.runInTransactionOrSavepoint(
_db,
transaction,
(savepoint) => _db.insertRow(scopedRow, transaction: savepoint),
);
} on DatabaseQueryException {
final owner = await _readDomainRowOwner(
insert.tableName,
insert.uuidRowId,
transaction,
);
// No row means the violation came from something other than the
// primary key (e.g. a racing unique value); preserve the failure.
if (!owner.exists) rethrow;
if (owner.scopeId != mergingScopeId) {
throw _ForeignScopeRowCollision(owner.scopeId);
}
// Same-scope recovery (a tracked row whose metadata was lost, or a
// race between this scope's own merges). scopeId is immutable and rows
// are only soft-deleted, so ownership observed here cannot change for
// the rest of the transaction and the update is safe.
await _db.updateRow(
patchedRow,
columns: patchedRow.table.managedColumns
scopedRow,
columns: scopedRow.table.managedColumns
.where((c) => data.containsKey(c.columnName))
.toList(),
transaction: transaction,
);
} on DatabaseUpdateRowException {
await _db.insertRow(patchedRow, transaction: transaction);
}
return insertedCrdtRow;
@ -579,6 +684,19 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
_MergeContext context,
Transaction transaction,
) async {
if (!await _domainRowOwnedByEffectiveScope(
operation: CrdtSyncViolationOperation.mergeInsert,
tableName: insert.tableName,
rowId: insert.uuidRowId,
context: context,
transaction: transaction,
metadataRowId: currentRow.id,
node: remoteNode,
hlc: insert.hlc,
)) {
return;
}
final (_, columnsByName) = _schema[insert.tableName]!;
final updatedValues = <String, Object?>{};
for (final MapEntry(key: columnName, value: schemaColumn)
@ -715,12 +833,133 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
return {
for (final MapEntry(key: columnName, value: value) in data.entries)
if (columnName != 'id' && columns.containsKey(columnName))
if (columnName != 'id' &&
columnName != 'scopeId' &&
columns.containsKey(columnName))
columnName: columns[columnName]!.columnType == ColumnType.uuid
? _uuidValueFromDatabase(value)
: value,
};
}
Future<({bool exists, int? scopeId})> _readDomainRowOwner(
String tableName,
UuidValue rowId,
Transaction transaction,
) async {
final values = await _readDomainColumnValues(
tableName,
{rowId},
['scopeId'],
transaction,
);
final row = values[rowId];
if (row == null) return (exists: false, scopeId: null);
return (exists: true, scopeId: row['scopeId'] as int?);
}
/// Returns the owning scope of [rowId] from the merge snapshot taken in
/// [_loadMergeContext], reading the domain row only if it was not captured
/// there (e.g. a row a concurrent operation added since the snapshot).
Future<_DomainOwner> _resolveDomainOwner(
_MergeContext context,
String tableName,
UuidValue rowId,
Transaction transaction,
) async {
final key = (tableName, rowId);
final cached = context.domainOwners[key];
if (cached != null) return cached;
final owner = await _readDomainRowOwner(tableName, rowId, transaction);
context.domainOwners[key] = owner;
return owner;
}
Future<bool> _domainRowOwnedByEffectiveScope({
required CrdtSyncViolationOperation operation,
required String tableName,
required UuidValue rowId,
required _MergeContext context,
required Transaction transaction,
int? metadataRowId,
CrdtNode? node,
Hlc? hlc,
}) async {
final owner = await _resolveDomainOwner(context, tableName, rowId, transaction);
final mergingScope = _getEffectiveScope(transaction);
if (owner.exists && owner.scopeId == mergingScope.id) return true;
if (!owner.exists) return false;
await _throwOwnershipCollision(
operation: operation,
tableName: tableName,
rowId: rowId,
owningScopeId: owner.scopeId,
incomingScope: mergingScope,
metadataRowId: metadataRowId,
node: node,
hlc: hlc,
transaction: transaction,
);
}
Future<Never> _throwOwnershipCollision({
required CrdtSyncViolationOperation operation,
required String tableName,
required UuidValue rowId,
required int? owningScopeId,
required CrdtScope incomingScope,
required int? metadataRowId,
required Transaction transaction,
CrdtNode? node,
Hlc? hlc,
}) async {
final now = DateTime.now().toUtc();
throw CrdtSyncIntegrityViolationException(
CrdtSyncIntegrityViolation(
type: CrdtSyncViolationType.ownershipCollision,
domainTableName: tableName,
uuidRowId: rowId,
ownerScopeUuid: await _scopeUuidForNormalizedId(owningScopeId, transaction),
incomingScopeUuid: incomingScope.uuidScopeId,
operation: operation,
uuidNodeId: node?.uuidNodeId,
crdtDataRowId: metadataRowId,
hlcDatetime: hlc?.datetime,
hlcCounter: hlc?.counter,
firstSeenAt: now,
lastSeenAt: now,
occurrences: 1,
),
);
}
Future<UuidValue?> _scopeUuidForNormalizedId(
int? scopeId,
Transaction transaction,
) async {
if (scopeId == null) return null;
final scope = await CrdtScope.db.findById(
_session,
scopeId,
transaction: transaction,
);
return scope?.uuidScopeId;
}
T _withScopeId<T extends TableRow>(T row, int scopeId) {
return (row as dynamic).copyWith(scopeId: scopeId) as T;
}
}
/// Signals that a merge insert lost a primary-key race to a row owned by
/// another scope, so the savepoint around the insert application must be
/// rolled back before the sync violation is recorded and thrown.
class _ForeignScopeRowCollision implements Exception {
_ForeignScopeRowCollision(this.owningScopeId);
final int? owningScopeId;
}
extension on DatabaseSerializationManager {
@ -728,7 +967,9 @@ extension on DatabaseSerializationManager {
T patchTableRow<T extends TableRow>(T row, Map<String, Object?> data) {
final rowJson = Map<String, dynamic>.from(row.toJson() as Map);
for (final column in row.table.managedColumns) {
if (column.columnName != 'id' && data.containsKey(column.columnName)) {
if (column.columnName != 'id' &&
column.columnName != 'scopeId' &&
data.containsKey(column.columnName)) {
rowJson[column.fieldName] = data[column.columnName];
}
}

17
packages/serverpod_offline_sync_client/lib/src/database/merge_utils/foreign_key_projector.dart
View file

@ -409,10 +409,10 @@ extension _CrdtForeignKeyProjector on CrdtMutationRecorder {
for (final tableName in _syncTableByName.keys) {
final (tableId, _) = _schema[tableName]!;
final userId = _getHlcManager(transaction).normalizedUserId;
final userId = _getHlcManager(transaction).normalizedScopeId;
final crdtRows = await CrdtDataRow.db.find(
_session,
where: (t) => t.userId.equals(userId) & t.tblId.equals(tableId),
where: (t) => t.scopeId.equals(userId) & t.tblId.equals(tableId),
include: CrdtDataRow.include(deleted: CrdtDataDeleted.include()),
orderBy: (t) => t.uuidRowId,
transaction: transaction,
@ -519,11 +519,11 @@ extension _CrdtForeignKeyProjector on CrdtMutationRecorder {
if (rowIds.isEmpty || columnNames.isEmpty) return {};
final (tableId, _) = _schema[tableName]!;
final userId = _getHlcManager(transaction).normalizedUserId;
final userId = _getHlcManager(transaction).normalizedScopeId;
final fields = await CrdtDataField.db.find(
_session,
where: (t) =>
t.row.userId.equals(userId) &
t.row.scopeId.equals(userId) &
t.row.tblId.equals(tableId) &
t.row.uuidRowId.inSet(rowIds) &
t.column.name.inSet(columnNames),
@ -970,6 +970,7 @@ extension _CrdtForeignKeyProjector on CrdtMutationRecorder {
fields: <_MergeFieldKey, CrdtDataField>{},
incomingFieldHlcs: <_MergeFieldKey, Hlc>{},
tombstones: <_MergeRowKey, CrdtDataDeleted>{},
domainOwners: <_MergeRowKey, _DomainOwner>{},
);
for (final MapEntry(key: rowKey, value: updates) in changedUpdatesByRow.entries) {
@ -1059,7 +1060,13 @@ extension _CrdtForeignKeyProjector on CrdtMutationRecorder {
) async {
final rowIds = await _findVisibleDomainRowIdsWhere(
tableName: edge.parentTableName,
predicates: [_domainColumnPredicate(edge.parentColumn, value)],
predicates: [
_domainColumnPredicate(
'scopeId',
_getHlcManager(transaction).normalizedScopeId,
),
_domainColumnPredicate(edge.parentColumn, value),
],
transaction: transaction,
);
return rowIds.isNotEmpty;

9
packages/serverpod_offline_sync_client/lib/src/database/merge_utils/unique_resolver.dart
View file

@ -259,7 +259,14 @@ class CrdtUniqueConflictResolver {
if (values.containsKey(column.columnName))
column.columnName: values[column.columnName],
};
if (columnValues.length != uniqueIndex.columns.length) continue;
if (uniqueIndex.scoped) {
columnValues['scopeId'] = _recorder
._getHlcManager(transaction)
.normalizedScopeId;
}
final expectedValueCount =
uniqueIndex.columns.length + (uniqueIndex.scoped ? 1 : 0);
if (columnValues.length != expectedValueCount) continue;
if (columnValues.values.any((value) => value == null)) continue;
final uniqueConflictIds = await _findVisibleRowsByUniqueValues(

141
packages/serverpod_offline_sync_client/lib/src/database/recorder.dart
View file

@ -2,14 +2,16 @@ import 'package:serverpod_database/serverpod_database.dart';
import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart';
import 'package:serverpod_serialization/serverpod_serialization.dart';
import '../crdt/exceptions.dart';
import '../crdt/extensions.dart';
import '../crdt/merge.dart';
import '../managers/hlc.dart';
import '../managers/user.dart';
import '../managers/scope.dart';
import '../protocol/protocol.dart';
import 'database.dart';
import 'schema.dart';
import 'session.dart';
import 'unique_index_utils.dart';
part 'merge.dart';
part 'merge_utils/foreign_key_projector.dart';
@ -36,6 +38,7 @@ typedef _ForeignKeyValueKey = (String tableName, String columnName, String value
typedef _UniqueIndexConflictRelease = ({
List<_UniqueColumnConflictRelease> columns,
bool scoped,
});
typedef _UniqueColumnConflictRelease = ({
@ -76,7 +79,7 @@ class CrdtMutationRecorder {
persistentUserId: persistentUserId,
);
late final CrdtUserManager _userManager = CrdtUserManager(_dbSession);
late final CrdtScopeManager _scopeManager = CrdtScopeManager(_dbSession);
final Map<UuidValue, HlcManager> _hlcManagers = {};
@ -162,11 +165,11 @@ class CrdtMutationRecorder {
/// Safe to call again after the database was wiped (e.g. test `tearDown`)
/// for in-memory schema ids to match new rows.
Future<void> initialize() async {
_userManager.clearCache();
_scopeManager.clearCache();
_hlcManagers.clear();
if (persistentUserId != null) {
await _userManager.getOrCreate(persistentUserId!);
await _scopeManager.getOrCreate(persistentUserId!);
}
final schemaRegistry = CrdtSchemaRegistry(_dbSession, syncTables: syncTables);
@ -200,7 +203,8 @@ class CrdtMutationRecorder {
late final Map<String, Set<String>> _syncedTableColumnNamesForMerge = {
for (final MapEntry(key: k, value: cols) in _columnsByTableAndName.entries)
if (_syncTableByName.containsKey(k)) k: cols.keys.toSet(),
if (_syncTableByName.containsKey(k))
k: cols.keys.where((columnName) => columnName != 'scopeId').toSet(),
};
/// Whether the given table is tracked by CRDT.
@ -215,9 +219,25 @@ class CrdtMutationRecorder {
return _syncTableByName.containsKey(tableName);
}
/// Returns the [CrdtUser] for the given user ID, creating it when needed.
Future<CrdtUser> getOrCreateUser(UuidValue userId) {
return _userManager.getOrCreate(userId);
/// 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;
/// Returns the user scoping CRDT visibility for queries, or null when no
/// user is associated with [transaction] and no persistent user exists.
CrdtScope? scopeForQueries(Transaction? transaction) {
if (transaction != null) {
final user = scopeForTransaction[transaction];
if (user != null) return user;
}
final userId = persistentUserId;
if (userId == null) return null;
return _scopeManager.getCached(userId);
}
/// Returns the [CrdtScope] for the given user ID, creating it when needed.
Future<CrdtScope> getOrCreateScope(UuidValue userId) {
return _scopeManager.getOrCreate(userId);
}
/// Records the latest acknowledged sync checkpoint for [otherNodeId].
@ -226,17 +246,17 @@ class CrdtMutationRecorder {
UuidValue otherNodeId,
Hlc syncedHlc,
) async {
final user = await _userManager.getOrCreate(userId);
final user = await _scopeManager.getOrCreate(userId);
await _db.transaction((transaction) async {
var node = await CrdtNode.db.findFirstRow(
_session,
where: (t) => t.userId.equals(user.id) & t.uuidNodeId.equals(otherNodeId),
where: (t) => t.scopeId.equals(user.id) & t.uuidNodeId.equals(otherNodeId),
transaction: transaction,
);
node ??= await CrdtNode.db.insertRow(
_session,
CrdtNode(userId: user.id!, uuidNodeId: otherNodeId),
CrdtNode(scopeId: user.id!, uuidNodeId: otherNodeId),
transaction: transaction,
);
@ -421,7 +441,7 @@ class CrdtMutationRecorder {
}) async {
if (foreignKeys.isEmpty) return null;
final userId = _getHlcManager(transaction).normalizedUserId;
final scopeId = _getHlcManager(transaction).normalizedScopeId;
final escapedChildTable = _escapeIdentifier(childTableName);
final whereRowIds = _sqlLiteralList(childRowIds);
@ -440,10 +460,14 @@ FROM "$escapedChildTable" c
LEFT JOIN "$parentTable" p
ON p."$parentColumn" = c."$childColumn"
LEFT JOIN "crdt_data_rows" r
ON r."userId" = $userId AND r."tblId" = $parentTableId AND r."uuidRowId" = p."id"
ON r."scopeId" = $scopeId AND r."tblId" = $parentTableId AND r."uuidRowId" = p."id"
WHERE c."id" IN ($whereRowIds)
AND c."$childColumn" IS NOT NULL
AND (p."id" IS NULL OR r."visibility" > $crdtRowLastVisibleVisibilityIndex)
AND (
p."id" IS NULL OR
p."scopeId" <> $scopeId OR
r."visibility" > $crdtRowLastVisibleVisibilityIndex
)
''';
});
@ -487,7 +511,7 @@ WHERE c."id" IN ($whereRowIds)
final hlc = hlcManager.increment();
return CrdtDataRow(
userId: hlcManager.normalizedUserId,
scopeId: hlcManager.normalizedScopeId,
tblId: tableId,
uuidRowId: rowId,
nodeId: hlcManager.normalizedNodeId,
@ -552,7 +576,7 @@ WHERE c."id" IN ($whereRowIds)
final table = updatedRows.first.table;
final updatedColumnList = (columns ?? table.managedColumns)
.where((c) => c.columnName != 'id')
.where((c) => c.columnName != 'id' && c.columnName != 'scopeId')
.toList();
if (updatedColumnList.isEmpty) return;
@ -871,11 +895,13 @@ WHERE c."id" IN ($whereRowIds)
CrdtDataRowInclude? include,
}) {
final (tableId, _) = _schema[tableName]!;
final userId = _getHlcManager(transaction).normalizedUserId;
final scopeId = _getHlcManager(transaction).normalizedScopeId;
return CrdtDataRow.db.find(
_session,
where: (t) =>
t.userId.equals(userId) & t.tblId.equals(tableId) & t.uuidRowId.inSet(rowIds),
t.scopeId.equals(scopeId) &
t.tblId.equals(tableId) &
t.uuidRowId.inSet(rowIds),
include: include,
transaction: transaction,
);
@ -979,9 +1005,7 @@ WHERE c."id" IN ($whereRowIds)
final uniqueIndexes = _uniqueIndexesForTable(tableDefinition);
for (final uniqueIndex in uniqueIndexes) {
final columnNames = uniqueIndex.columns
.map((column) => column.columnName)
.toList();
final columnNames = uniqueIndex.columnNames.toList();
final valuesByRowId = await _readDomainColumnValues(
tableName,
@ -1025,7 +1049,10 @@ WHERE c."id" IN ($whereRowIds)
) {
return _uniqueIndexesByTableName.putIfAbsent(
tableDefinition.name,
() => _syncableUniqueIndexesForTable(tableDefinition),
() => _syncableUniqueIndexesForTable(
tableDefinition,
_syncTableByName.keys.toSet(),
),
);
}
@ -1048,13 +1075,13 @@ WHERE c."id" IN ($whereRowIds)
required Transaction transaction,
}) async {
final (tableId, _) = _schema[tableName]!;
final userId = _getHlcManager(transaction).normalizedUserId;
final scopeId = _getHlcManager(transaction).normalizedScopeId;
final result = await _db.unsafeQuery(
'''
SELECT d."id"
FROM "${_escapeIdentifier(tableName)}" d
LEFT JOIN "crdt_data_rows" r
ON r."userId" = $userId AND r."tblId" = $tableId AND r."uuidRowId" = d."id"
ON r."scopeId" = $scopeId AND r."tblId" = $tableId AND r."uuidRowId" = d."id"
WHERE (${predicates.join(') AND (')})
AND (r."id" IS NULL OR r."visibility" <= $crdtRowLastVisibleVisibilityIndex)
''',
@ -1170,20 +1197,18 @@ WHERE "id" IN (${_sqlLiteralList(rowIds)})
}
HlcManager _getHlcManager(Transaction transaction) {
final user = _getEffectiveUser(transaction);
final user = _getEffectiveScope(transaction);
return _hlcManagers.putIfAbsent(
user.uuidUserId,
() => HlcManager.forUser(user),
user.uuidScopeId,
() => HlcManager.forScope(user),
);
}
CrdtUser _getEffectiveUser(Transaction transaction) {
final user = userForTransaction[transaction];
if (user != null) return user;
if (persistentUserId == null) {
throw StateError('No user ID found for transaction or persistent user ID.');
}
return _userManager.getCached(persistentUserId!);
CrdtScope _getEffectiveScope(Transaction transaction) {
return scopeForQueries(transaction) ??
(throw StateError(
'No user ID found for transaction or persistent user ID.',
));
}
}
@ -1197,6 +1222,7 @@ int _nextClFlag(CrdtDataDeleted? current, bool isDeleted) {
List<_UniqueIndexConflictRelease> _syncableUniqueIndexesForTable(
TableDefinition table,
Set<String> syncTableNames,
) {
final uniqueIndexes = table.indexes
.where(
@ -1205,26 +1231,51 @@ List<_UniqueIndexConflictRelease> _syncableUniqueIndexesForTable(
!index.isPrimary &&
index.elements.every(
(element) => element.type == IndexElementDefinitionType.column,
),
) &&
(_isScopedUniqueIndex(index) ||
isCrdtAllowedForeignKeyOnlyUniqueIndex(
table,
index,
syncTableNames,
)),
)
.toList();
final columnsByName = {for (final column in table.columns) column.name: column};
return [
for (final index in uniqueIndexes)
(
columns: [
for (final columnName in index.elements.map((element) => element.definition))
_uniqueConflictReleaseForColumn(
table,
columnsByName[columnName],
columnName,
),
],
),
_uniqueIndexConflictReleaseForIndex(table, columnsByName, index),
];
}
bool _isScopedUniqueIndex(IndexDefinition index) {
return index.elements.any(
(element) =>
element.type == IndexElementDefinitionType.column &&
element.definition == 'scopeId',
);
}
_UniqueIndexConflictRelease _uniqueIndexConflictReleaseForIndex(
TableDefinition table,
Map<String, ColumnDefinition> columnsByName,
IndexDefinition index,
) {
final columnNames = index.elements.map((element) => element.definition).toList();
return (
scoped: columnNames.contains('scopeId'),
columns: [
for (final columnName in columnNames)
if (columnName != 'scopeId')
_uniqueConflictReleaseForColumn(
table,
columnsByName[columnName],
columnName,
),
],
);
}
_UniqueColumnConflictRelease _uniqueConflictReleaseForColumn(
TableDefinition table,
ColumnDefinition? column,

96
packages/serverpod_offline_sync_client/lib/src/database/schema.dart
View file

@ -2,11 +2,16 @@ import 'package:serverpod_database/serverpod_database.dart';
import 'package:uuid/uuid.dart';
import '../protocol/protocol.dart';
import 'unique_index_utils.dart';
/// Manages the CRDT schema for a database.
class CrdtSchemaRegistry {
/// Creates a new instance of [CrdtSchemaRegistry].
CrdtSchemaRegistry(this._session, {required this.syncTables}) {
CrdtSchemaRegistry(
this._session, {
required this.syncTables,
Iterable<TableDefinition>? tableDefinitions,
}) {
final tablesWithoutUuidPk = syncTables.where(
(table) => table.id is! Column<UuidValue>,
);
@ -17,6 +22,47 @@ class CrdtSchemaRegistry {
'${tablesWithoutUuidPk.map((t) => '"${t.tableName}"').join(', ')}',
);
}
final tablesWithoutScopeId = syncTables.where((table) {
final scopeColumn = _scopeIdColumn(table);
return scopeColumn == null || scopeColumn is! Column<int>;
}).toList();
if (tablesWithoutScopeId.isNotEmpty) {
throw StateError(
'CRDT can only synchronize tables with a nullable int scopeId column, '
'but ${tablesWithoutScopeId.length} table(s) are missing it or declare '
'it with the wrong type: '
'${tablesWithoutScopeId.map((t) => '"${t.tableName}"').join(', ')}\n\n'
'Add this field to every synced model:\n'
'fields:\n'
' id: UuidValue?, defaultPersist=random_v7\n'
' ### Owner scope of this row. Maintained by the CRDT sync layer.\n'
' scopeId: int?\n\n'
'If the column is declared with scope=serverOnly, remove the scope; '
'the column must exist on every database.',
);
}
final tableDefinitionsByName = {
for (final tableDefinition
in tableDefinitions ??
_session.db.serializationManager.getTargetTableDefinitions())
tableDefinition.name: tableDefinition,
};
final globalUniqueIndexes = _globalUniqueIndexViolations(
syncTables,
tableDefinitionsByName,
);
if (globalUniqueIndexes.isNotEmpty) {
throw StateError(
'CRDT can only synchronize tables with per-scope unique indexes, but '
'${globalUniqueIndexes.length} unique index(es) do not include scopeId: '
'${globalUniqueIndexes.join(', ')}. '
'Add scopeId to the unique index fields. '
'The only allowed global unique indexes are foreign-key-only indexes '
'whose target tables are also synchronized.',
);
}
}
final DatabaseSession _session;
@ -27,7 +73,8 @@ class CrdtSchemaRegistry {
late final _columnsPerTableName = {
for (final table in syncTables)
table.tableName: [
for (final column in table.columns) column.columnName,
for (final column in table.columns)
if (column.columnName != 'scopeId') column.columnName,
],
};
@ -118,3 +165,48 @@ class CrdtSchemaRegistry {
return foundColumnRows;
}
}
Column? _scopeIdColumn(Table table) {
for (final column in table.columns) {
if (column.columnName == 'scopeId') return column;
}
return null;
}
List<String> _globalUniqueIndexViolations(
List<Table> syncTables,
Map<String, TableDefinition> tableDefinitionsByName,
) {
final syncTableNames = {for (final table in syncTables) table.tableName};
return [
for (final tableName in syncTableNames)
for (final index
in tableDefinitionsByName[tableName]?.indexes ?? const <IndexDefinition>[])
if (_isForbiddenGlobalUniqueIndex(
tableDefinitionsByName[tableName]!,
index,
syncTableNames,
))
'$tableName.${index.indexName}',
];
}
bool _isForbiddenGlobalUniqueIndex(
TableDefinition tableDefinition,
IndexDefinition index,
Set<String> syncTableNames,
) {
if (!index.isUnique || index.isPrimary) return false;
final includesScopeId = index.elements.any(
(element) =>
element.type == IndexElementDefinitionType.column &&
element.definition == 'scopeId',
);
if (includesScopeId) return false;
return !isCrdtAllowedForeignKeyOnlyUniqueIndex(
tableDefinition,
index,
syncTableNames,
);
}

172
packages/serverpod_offline_sync_client/lib/src/database/scope.dart Normal file
View file

@ -0,0 +1,172 @@
part of 'database.dart';
extension _CrdtDatabaseScope on CrdtDatabase {
({
List<T> rows,
Set<int> stampedIndexes,
Set<Object> explicitRowIds,
})
_prepareRowsForInsert<T extends TableRow>(
List<T> rows,
Transaction transaction,
) {
if (!_recorder.isCrdtTracked<T>(rows.first.table)) {
return (rows: rows, stampedIndexes: const {}, explicitRowIds: const {});
}
final scope = _requireEffectiveScope(transaction);
final effectiveScopeId = scope.id!;
final databaseRows = <T>[];
final stampedIndexes = <int>{};
final explicitRowIds = <Object>{};
for (final (index, row) in rows.indexed) {
final rowScopeId = _readScopeId(row);
if (rowScopeId == null) {
databaseRows.add(_copyWithScopeId(row, effectiveScopeId));
stampedIndexes.add(index);
continue;
}
if (rowScopeId != effectiveScopeId) {
throw StateError(
'Cannot write ${row.table.tableName} row ${row.id} with scopeId '
'$rowScopeId while acting in scope $effectiveScopeId.',
);
}
databaseRows.add(row);
final rowId = row.id;
if (rowId != null) explicitRowIds.add(rowId as Object);
}
return (
rows: databaseRows,
stampedIndexes: stampedIndexes,
explicitRowIds: explicitRowIds,
);
}
bool _shouldStripReturnedScopeId<T extends TableRow>(
T row,
Transaction transaction,
) {
if (!_recorder.isCrdtTracked<T>(row.table)) return false;
final rowScopeId = _readScopeId(row);
if (rowScopeId == null) return true;
final effectiveScopeId = _requireEffectiveScope(transaction).id!;
if (rowScopeId != effectiveScopeId) {
throw StateError(
'Cannot write ${row.table.tableName} row ${row.id} with scopeId '
'$rowScopeId while acting in scope $effectiveScopeId.',
);
}
return false;
}
void _assertNoScopeIdColumnValues<T extends TableRow>(
List<ColumnValue> columnValues,
) {
for (final columnValue in columnValues) {
if (columnValue.column.columnName == 'scopeId') {
final table = serializationManager.getTableForType(T)?.tableName ?? 'unknown';
throw StateError(
'scopeId is immutable and owned by the CRDT sync layer for $table.',
);
}
}
}
CrdtScope _requireEffectiveScope(Transaction transaction) {
return _recorder.scopeForQueries(transaction) ??
(throw StateError(
'A scope ID is required for CRDT writes without a persistent user.',
));
}
int? _readScopeId(TableRow row) => (row as dynamic).scopeId as int?;
T _copyWithScopeId<T extends TableRow>(T row, int scopeId) {
return (row as dynamic).copyWith(scopeId: scopeId) as T;
}
void _stripStampedRows<T extends TableRow>(
List<T> rows,
({
List<T> rows,
Set<int> stampedIndexes,
Set<Object> explicitRowIds,
})
prepared,
) {
if (rows.length == prepared.rows.length) {
for (final (index, row) in rows.indexed) {
if (prepared.stampedIndexes.contains(index)) _stripScopeId(row);
}
return;
}
// `ignoreConflicts` dropped rows, so positions no longer align with the
// input. Keep the value only on rows whose id was caller-provided on an
// explicit (asserted) input row; everything else is treated as stamped.
for (final row in rows) {
final rowId = row.id;
if (rowId == null || !prepared.explicitRowIds.contains(rowId)) {
_stripScopeId(row);
}
}
}
void _stripScopeId(TableRow row) {
(row as dynamic).scopeId = null;
}
List<T> _stripScopeIdFromScopedRead<T extends TableRow>(
List<T> rows,
Include? include,
Transaction? transaction,
) {
if (rows.isEmpty) return rows;
if (!_recorder.isCrdtTracked<T>(rows.first.table)) return rows;
if (_recorder.scopeForQueries(transaction) == null) return rows;
final hasNestedRows =
include != null && include.includes.values.any((inc) => inc != null);
if (!hasNestedRows) {
// Hot path: delegate-returned rows are package-owned, so the root
// field is stripped in place with no allocation.
rows.forEach(_stripScopeId);
return rows;
}
// Nested models are only reachable through generated fields, which
// cannot be accessed generically by name, so include graphs pay a
// serialization round-trip per row to strip recursively.
return [
for (final row in rows) _copyWithScopeIdsStripped(row),
];
}
T _copyWithScopeIdsStripped<T extends TableRow>(T row) {
final json = Map<String, dynamic>.from(row.toJson() as Map);
_stripScopeIdsFromJson(json);
return serializationManager.deserialize<T>(json);
}
void _stripScopeIdsFromJson(Object? value) {
if (value is Map) {
if (value.containsKey('scopeId')) {
value['scopeId'] = null;
}
value.values.forEach(_stripScopeIdsFromJson);
return;
}
if (value is Iterable) {
value.forEach(_stripScopeIdsFromJson);
}
}
}

132
packages/serverpod_offline_sync_client/lib/src/database/tombstone.dart
View file

@ -4,12 +4,33 @@ import 'package:serverpod_database/serverpod_database.dart';
import '../crdt/extensions.dart';
import '../protocol/protocol.dart';
/// Resolves the local [CrdtSchemaTable] id for a domain table name. Returns
/// null when the table is not registered for CRDT synchronization.
typedef CrdtTableIdResolver = int? Function(String tableName);
/// Resolves the [CrdtScope] id scoping the current query. Returns null when
/// no scope is available (e.g. server-side reads outside `transactionForUser`
/// without a persistent user), in which case the query is an admin read: rows
/// are not isolated and a row is hidden only when its owning scope hides it.
typedef CrdtScopeIdResolver = int? Function();
/// Merges [where] with CRDT visibility predicates and mutates [include] in place.
///
/// This method ensures that no rows with a hidden [CrdtDataRow.visibility] are
/// returned for queries with [where] clauses or [include] graphs. If [include]
/// is null, only the root table and [where] are merged.
///
/// Each predicate only considers [CrdtDataRow]s recorded for the queried table
/// ([tableIdForName]) and, when a scope is available ([scopeId]), isolates the
/// query to that scope's rows. This prevents a tombstone recorded for another
/// table or scope from masking an unrelated row that reuses the same UUID.
/// Without a scope (admin reads), rows are not isolated and a row stays
/// visible as long as its owning scope still sees it.
///
/// Both resolvers are only invoked when a predicate is actually built (i.e.
/// for synced tables with UUID primary keys), so queries on CRDT-internal
/// tables never touch the recorder state.
///
/// We do **not** add a root predicate for keys whose nested include is an
/// [IncludeList] - visibility filtering for that path is already handled via the
/// list subquery [where] (and many-relations are not the same join shape as a
@ -18,14 +39,21 @@ import '../protocol/protocol.dart';
Expression? mergeWhereWithTombstone<T extends TableRow>(
DatabaseSerializationManager serializationManager,
Expression? where,
Include? include,
) {
final includeObjectPredicates = _walkIncludeGraphForTombstone(include, null);
Include? include, {
required CrdtTableIdResolver tableIdForName,
required CrdtScopeIdResolver scopeId,
}) {
final includeObjectPredicates = _walkIncludeGraphForTombstone(
include,
null,
tableIdForName: tableIdForName,
scopeId: scopeId,
);
final rootTable = serializationManager.getTableForType(T);
var merged = _mergeWhereOptional(
where,
rootTable?.whereVisibleOnCrdtRow,
rootTable?.whereVisibleOnCrdtRow(tableIdForName, scopeId),
);
merged = _mergeWhereOptional(merged, includeObjectPredicates);
return merged;
@ -35,16 +63,23 @@ Expression? mergeWhereWithTombstone<T extends TableRow>(
/// `AND` into the main query for [IncludeObject] joins.
Expression? _walkIncludeGraphForTombstone(
Include? inc,
Expression? includeObjectPredicates,
) {
Expression? includeObjectPredicates, {
required CrdtTableIdResolver tableIdForName,
required CrdtScopeIdResolver scopeId,
}) {
if (inc == null) return includeObjectPredicates;
if (inc is IncludeList) {
// List relation: filter inside the list subquery only.
inc.where = _mergeWhereOptional(
inc.where,
inc.table.whereVisibleOnCrdtRow,
inc.table.whereVisibleOnCrdtRow(tableIdForName, scopeId),
);
return _walkIncludeGraphForTombstone(
inc.include,
includeObjectPredicates,
tableIdForName: tableIdForName,
scopeId: scopeId,
);
return _walkIncludeGraphForTombstone(inc.include, includeObjectPredicates);
}
var acc = includeObjectPredicates;
final obj = inc;
@ -53,7 +88,12 @@ Expression? _walkIncludeGraphForTombstone(
final nested = entry.value;
if (nested is IncludeList) {
// e.g. one-to-many: visibility applies via IncludeList.where, not the main WHERE.
acc = _walkIncludeGraphForTombstone(nested, acc);
acc = _walkIncludeGraphForTombstone(
nested,
acc,
tableIdForName: tableIdForName,
scopeId: scopeId,
);
continue;
}
if (nested is IncludeObject) {
@ -62,10 +102,15 @@ Expression? _walkIncludeGraphForTombstone(
if (childTable != null) {
acc = _mergeWhereOptional(
acc,
childTable.whereVisibleOnCrdtRow,
childTable.whereVisibleOnCrdtRow(tableIdForName, scopeId),
);
}
acc = _walkIncludeGraphForTombstone(nested, acc);
acc = _walkIncludeGraphForTombstone(
nested,
acc,
tableIdForName: tableIdForName,
scopeId: scopeId,
);
}
}
return acc;
@ -81,25 +126,56 @@ extension on Table {
/// Creates a predicate that filters out hidden rows for the given [Table]
/// using materialized [CrdtDataRow.visibility] metadata.
///
/// Join path:
/// - UserModel.[id] = [CrdtDataRow.uuidRowId].
/// Returns null when [Table] does not use a UUID primary key or is not
/// registered for CRDT synchronization. Keeps rows when the id is null
/// (unmatched [IncludeObject] joins) or no hidden CRDT row matches.
///
/// Returns null when [Table] does not use a UUID primary key. Keeps rows when
/// there is no CRDT row or [CrdtDataRow.visibility] is visible (LEFT JOIN
/// null-safe).
Expression? get whereVisibleOnCrdtRow {
/// With a scope, rows are isolated to it (`scopeId = <scope>`) and checked
/// against the scope's tombstones with a correlated `NOT EXISTS` probe
/// covered by the (scopeId, tblId, uuidRowId) unique index. Without a
/// scope (admin reads), there is no isolation filter and the probe is
/// keyed by the row's own `scopeId` column instead — covered by the same
/// index — so a row stays visible while its owning scope sees it.
Expression? whereVisibleOnCrdtRow(
CrdtTableIdResolver tableIdForName,
CrdtScopeIdResolver scopeId,
) {
if (id is! ColumnUuid) return null;
final crdtRowTable = createRelationTable<CrdtDataRowTable>(
relationFieldName: '${tableName}_crdt_row',
field: id,
foreignField: CrdtDataRow.t.uuidRowId,
tableRelation: tableRelation,
createTable: (foreignTableRelation) =>
CrdtDataRowTable(tableRelation: foreignTableRelation),
final tableId = tableIdForName(tableName);
if (tableId == null) return null;
// Synced tables are validated to carry the column at initialize(); a
// missing column here must fail closed, never skip the predicate.
final scopeColumn =
scopeIdColumn ??
(throw StateError(
'Synced table "$tableName" has no int scopeId column; '
'CrdtSchemaRegistry validation should have rejected it.',
));
final crdtRow = CrdtDataRow.t;
final effectiveScopeId = scopeId();
final notExistsExpr = Expression(
'NOT EXISTS '
'(SELECT 1 FROM "${crdtRow.tableName}" '
'WHERE ${crdtRow.scopeId} = ${effectiveScopeId ?? scopeColumn} '
'AND ${crdtRow.tblId} = $tableId '
'AND ${crdtRow.uuidRowId} = $id '
'AND ${crdtRow.visibility} > $crdtRowLastVisibleVisibilityIndex)',
);
return (crdtRowTable.id.equals(null)) |
Expression(
'${crdtRowTable.visibility} <= $crdtRowLastVisibleVisibilityIndex',
);
return id.equals(null) |
((effectiveScopeId != null)
? (scopeColumn.equals(effectiveScopeId) & notExistsExpr)
: notExistsExpr);
}
ColumnInt? get scopeIdColumn {
for (final column in columns) {
if (column.columnName == 'scopeId' && column is ColumnInt) {
return column;
}
}
return null;
}
}

31
packages/serverpod_offline_sync_client/lib/src/database/unique_index_utils.dart Normal file
View file

@ -0,0 +1,31 @@
import 'package:meta/meta.dart';
import 'package:serverpod_database/serverpod_database.dart';
@internal
bool isCrdtAllowedForeignKeyOnlyUniqueIndex(
TableDefinition tableDefinition,
IndexDefinition index,
Set<String> syncTableNames,
) {
final indexColumns = <String>[];
for (final element in index.elements) {
if (element.type != IndexElementDefinitionType.column) return false;
indexColumns.add(element.definition);
}
if (indexColumns.isEmpty || indexColumns.length != indexColumns.toSet().length) {
return false;
}
final indexColumnSet = indexColumns.toSet();
final coveredForeignKeyColumns = <String>{};
for (final foreignKey in tableDefinition.foreignKeys) {
if (!syncTableNames.contains(foreignKey.referenceTable)) continue;
final foreignKeyColumns = foreignKey.columns.toSet();
if (foreignKeyColumns.isEmpty) continue;
if (indexColumnSet.containsAll(foreignKeyColumns)) {
coveredForeignKeyColumns.addAll(foreignKeyColumns);
}
}
return indexColumns.every(coveredForeignKeyColumns.contains);
}

26
packages/serverpod_offline_sync_client/lib/src/managers/hlc.dart
View file

@ -7,27 +7,27 @@ import '../protocol/protocol.dart';
/// for CRDT operations.
class HlcManager {
HlcManager._(
this.uuidUserId,
this.normalizedUserId,
this.uuidScopeId,
this.normalizedScopeId,
this.normalizedNodeId,
this.lastHlc,
);
/// Creates a new [HlcManager] for the current node of [user].
factory HlcManager.forUser(CrdtUser user) {
/// Creates a new [HlcManager] for the current node of [scope].
factory HlcManager.forScope(CrdtScope scope) {
return HlcManager._(
user.uuidUserId,
user.id!,
user.currentNodeId!,
user.currentNode!.lastReceivedHlc ?? Hlc.zero(user.currentNode!.uuidNodeId),
scope.uuidScopeId,
scope.id!,
scope.currentNodeId!,
scope.currentNode!.lastReceivedHlc ?? Hlc.zero(scope.currentNode!.uuidNodeId),
);
}
/// The UUID of the user this manager is for.
final UuidValue uuidUserId;
/// The UUID of the scope this manager is for.
final UuidValue uuidScopeId;
/// The normalized user ID of the user this manager is for.
final int normalizedUserId;
/// The normalized scope ID of the scope this manager is for.
final int normalizedScopeId;
/// The node ID of the current node.
UuidValue get uuidNodeId => lastHlc.nodeId;
@ -52,7 +52,7 @@ class HlcManager {
/// Converts this manager state to the persisted current-node model.
CrdtNode getNode() => CrdtNode(
id: normalizedNodeId,
userId: normalizedUserId,
scopeId: normalizedScopeId,
uuidNodeId: uuidNodeId,
lastReceivedHlc: lastHlc,
);

76
packages/serverpod_offline_sync_client/lib/src/managers/scope.dart Normal file
View file

@ -0,0 +1,76 @@
import 'package:serverpod_database/serverpod_database.dart';
import 'package:uuid/uuid.dart';
import '../protocol/protocol.dart';
/// A manager for [CrdtScope] instances.
class CrdtScopeManager {
/// Creates a [CrdtScopeManager] bound to a database session.
CrdtScopeManager(this._session);
final DatabaseSession _session;
final Map<UuidValue, CrdtScope> _instances = {};
/// Returns the cached [CrdtScope] for the given scope ID.
CrdtScope getCached(UuidValue uuidScopeId) =>
_instances[uuidScopeId] ??
(throw StateError(
'Scope $uuidScopeId not found in cache. '
'Ensure CrdtScopeManager.getOrCreate() is called before getCached().',
));
/// Clears the in-memory cache so state is reloaded from the store.
void clearCache() {
_instances.clear();
}
/// Returns the [CrdtScope] for the given scope ID.
///
/// Will create a new [CrdtScope] if no scope is found.
Future<CrdtScope> getOrCreate(UuidValue uuidScopeId) async {
return _instances[uuidScopeId] ??= await _session.db.transaction(
(transaction) => _getOrCreate(uuidScopeId, transaction),
);
}
Future<CrdtScope> _getOrCreate(
UuidValue uuidScopeId,
Transaction transaction,
) async {
var scope = await CrdtScope.db.findFirstRow(
_session,
where: (t) => t.uuidScopeId.equals(uuidScopeId),
include: CrdtScope.include(currentNode: CrdtNode.include()),
transaction: transaction,
);
scope ??= await CrdtScope.db.insertRow(
_session,
CrdtScope(uuidScopeId: uuidScopeId),
transaction: transaction,
);
if (scope.currentNode != null) {
return scope;
}
final currentNode = await CrdtNode.db.insertRow(
_session,
CrdtNode(scopeId: scope.id!),
transaction: transaction,
);
await CrdtScope.db.attachRow.currentNode(
_session,
scope,
currentNode,
transaction: transaction,
);
return scope.copyWith(
currentNodeId: currentNode.id,
currentNode: currentNode,
);
}
}

76
packages/serverpod_offline_sync_client/lib/src/managers/user.dart
View file

@ -1,76 +0,0 @@
import 'package:serverpod_database/serverpod_database.dart';
import 'package:uuid/uuid.dart';
import '../protocol/protocol.dart';
/// A manager for [CrdtUser] instances.
class CrdtUserManager {
/// Creates a [CrdtUserManager] bound to a database session.
CrdtUserManager(this._session);
final DatabaseSession _session;
final Map<UuidValue, CrdtUser> _instances = {};
/// Returns the cached [CrdtUser] for the given user ID.
CrdtUser getCached(UuidValue uuidUserId) =>
_instances[uuidUserId] ??
(throw StateError(
'User $uuidUserId not found in cache. '
'Ensure CrdtUserManager.getOrCreate() is called before getCached().',
));
/// Clears the in-memory cache so state is reloaded from the store.
void clearCache() {
_instances.clear();
}
/// Returns the [CrdtUser] for the given user ID.
///
/// Will create a new [CrdtUser] if no user is found.
Future<CrdtUser> getOrCreate(UuidValue uuidUserId) async {
return _instances[uuidUserId] ??= await _session.db.transaction(
(transaction) => _getOrCreate(uuidUserId, transaction),
);
}
Future<CrdtUser> _getOrCreate(
UuidValue uuidUserId,
Transaction transaction,
) async {
var user = await CrdtUser.db.findFirstRow(
_session,
where: (t) => t.uuidUserId.equals(uuidUserId),
include: CrdtUser.include(currentNode: CrdtNode.include()),
transaction: transaction,
);
user ??= await CrdtUser.db.insertRow(
_session,
CrdtUser(uuidUserId: uuidUserId),
transaction: transaction,
);
if (user.currentNode != null) {
return user;
}
final currentNode = await CrdtNode.db.insertRow(
_session,
CrdtNode(userId: user.id!),
transaction: transaction,
);
await CrdtUser.db.attachRow.currentNode(
_session,
user,
currentNode,
transaction: transaction,
);
return user.copyWith(
currentNodeId: currentNode.id,
currentNode: currentNode,
);
}
}