refactor: Improve the code separation (#109)
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Marcelo Mendonça Soares
GitHub
parent
6b22199272
commit
1ddfef2988
16 changed files with 1697 additions and 1432 deletions
|
|
@ -67,6 +67,23 @@ extension CrdtDataDeletedReasonExtension on CrdtDataDeletedReason {
|
|||
/// Whether this tombstone reason is synced across replicas. All reasons that
|
||||
/// start with 'user' are synced.
|
||||
bool get isSynced => name.startsWith('user');
|
||||
|
||||
/// The row visibility that corresponds to this tombstone reason.
|
||||
CrdtDataRowVisibility toVisibility({required bool isDeleted}) {
|
||||
if (!isDeleted) {
|
||||
return switch (this) {
|
||||
CrdtDataDeletedReason.userReinsert => CrdtDataRowVisibility.userReinsert,
|
||||
_ => CrdtDataRowVisibility.userInsert,
|
||||
};
|
||||
}
|
||||
|
||||
return switch (this) {
|
||||
CrdtDataDeletedReason.userDelete => CrdtDataRowVisibility.userDelete,
|
||||
CrdtDataDeletedReason.userCascadeDelete =>
|
||||
CrdtDataRowVisibility.userCascadeDelete,
|
||||
_ => CrdtDataRowVisibility.userDelete,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/// CRDT write capability for a scope role.
|
||||
|
|
|
|||
|
|
@ -19,16 +19,21 @@ enum CrdtSyncPeerMode {
|
|||
|
||||
/// De-duplicates [scopeIds] and sorts them by UUID string so both peers iterate
|
||||
/// scopes in the same deterministic order.
|
||||
List<UuidValue> _sortedUniqueScopeIds(Iterable<UuidValue> scopeIds) {
|
||||
final byUuid = {for (final scopeId in scopeIds) scopeId.uuid: scopeId};
|
||||
return byUuid.values.toList()..sort((a, b) => a.uuid.compareTo(b.uuid));
|
||||
}
|
||||
List<UuidValue> _sortedUniqueScopeIds(Iterable<UuidValue> scopeIds) =>
|
||||
_sortedUniqueByUuid(scopeIds, (scopeId) => scopeId);
|
||||
|
||||
/// De-duplicates [grants] by scope and sorts them by scope UUID string.
|
||||
List<CrdtScopeGrant> _sortedUniqueGrants(Iterable<CrdtScopeGrant> grants) {
|
||||
final byUuid = {for (final grant in grants) grant.uuidScopeId.uuid: grant};
|
||||
List<CrdtScopeGrant> _sortedUniqueGrants(Iterable<CrdtScopeGrant> grants) =>
|
||||
_sortedUniqueByUuid(grants, (grant) => grant.uuidScopeId);
|
||||
|
||||
/// De-duplicates [items] and sorts them by the UUID returned by [uuidOf].
|
||||
List<T> _sortedUniqueByUuid<T>(
|
||||
Iterable<T> items,
|
||||
UuidValue Function(T) uuidOf,
|
||||
) {
|
||||
final byUuid = {for (final item in items) uuidOf(item).uuid: item};
|
||||
return byUuid.values.toList()
|
||||
..sort((a, b) => a.uuidScopeId.uuid.compareTo(b.uuidScopeId.uuid));
|
||||
..sort((a, b) => uuidOf(a).uuid.compareTo(uuidOf(b).uuid));
|
||||
}
|
||||
|
||||
/// Whether the grant list [a] last announced equals the current list [b].
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import 'package:serverpod_database/serverpod_database.dart';
|
|||
import 'package:uuid/uuid.dart';
|
||||
|
||||
import '../database/database.dart';
|
||||
import '../database/merge_utils/database_helpers.dart';
|
||||
import '../database/recorder.dart';
|
||||
import '../generated/protocol.dart';
|
||||
import '../hlc/hlc.dart';
|
||||
|
|
@ -828,11 +829,11 @@ class CrdtSync {
|
|||
DomainRowOwnerCache ownerCache,
|
||||
) async {
|
||||
final cols = table.columns
|
||||
.map((column) => '"${_escapeIdentifier(column.columnName)}"')
|
||||
.map((column) => '"${column.columnName.escapeIdentifier()}"')
|
||||
.join(', ');
|
||||
final encodedRowId = ValueEncoder.instance.convert(rowId);
|
||||
final encodedScopeId = ValueEncoder.instance.convert(scopeId);
|
||||
final escapedTableName = _escapeIdentifier(tableName);
|
||||
final encodedRowId = rowId.sqlLiteral();
|
||||
final encodedScopeId = scopeId.sqlLiteral();
|
||||
final escapedTableName = tableName.escapeIdentifier();
|
||||
final result = await session.db.unsafeQuery(
|
||||
'SELECT $cols FROM "$escapedTableName" '
|
||||
'WHERE "id" = $encodedRowId AND "scopeId" = $encodedScopeId '
|
||||
|
|
@ -886,13 +887,13 @@ class CrdtSync {
|
|||
);
|
||||
}
|
||||
|
||||
final encodedValue = ValueEncoder.instance.convert(rowId);
|
||||
final encodedScopeId = ValueEncoder.instance.convert(scopeId);
|
||||
final escapedTableName = _escapeIdentifier(tableName);
|
||||
final encodedRowId = rowId.sqlLiteral();
|
||||
final encodedScopeId = scopeId.sqlLiteral();
|
||||
final escapedTableName = tableName.escapeIdentifier();
|
||||
final result = await session.db.unsafeQuery(
|
||||
'SELECT "${_escapeIdentifier(columnName)}" '
|
||||
'SELECT "${columnName.escapeIdentifier()}" '
|
||||
'FROM "$escapedTableName" '
|
||||
'WHERE "id" = $encodedValue AND "scopeId" = $encodedScopeId '
|
||||
'WHERE "id" = $encodedRowId AND "scopeId" = $encodedScopeId '
|
||||
'LIMIT 1',
|
||||
);
|
||||
if (result.isNotEmpty) {
|
||||
|
|
@ -917,11 +918,11 @@ class CrdtSync {
|
|||
final cached = ownerCache[(tableName, rowId)];
|
||||
if (cached != null) return cached;
|
||||
|
||||
final encodedValue = ValueEncoder.instance.convert(rowId);
|
||||
final escapedTableName = _escapeIdentifier(tableName);
|
||||
final encodedRowId = rowId.sqlLiteral();
|
||||
final escapedTableName = tableName.escapeIdentifier();
|
||||
final result = await session.db.unsafeQuery(
|
||||
'SELECT "scopeId" FROM "$escapedTableName" '
|
||||
'WHERE "id" = $encodedValue '
|
||||
'WHERE "id" = $encodedRowId '
|
||||
'LIMIT 1',
|
||||
);
|
||||
final owner = result.isEmpty
|
||||
|
|
@ -1049,8 +1050,6 @@ class CrdtSync {
|
|||
return withoutNullable.split(':').last;
|
||||
}
|
||||
|
||||
String _escapeIdentifier(String identifier) => identifier.replaceAll('"', '""');
|
||||
|
||||
static String _computeCanonicalSyncTablesSignature(
|
||||
List<Table> syncTables, {
|
||||
required List<TableDefinition> tableDefinitions,
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ import '../crdt/scope_membership.dart';
|
|||
import '../crdt/sync.dart';
|
||||
import '../generated/protocol.dart';
|
||||
import '../hlc/hlc.dart';
|
||||
import 'merge_utils/database_helpers.dart';
|
||||
import 'recorder.dart';
|
||||
import 'session.dart';
|
||||
import 'tombstone.dart';
|
||||
|
|
@ -637,8 +638,7 @@ class CrdtDatabase implements Database {
|
|||
List<Column>? columns,
|
||||
}) async {
|
||||
final values = row.toJsonForDatabase() as Map<String, dynamic>;
|
||||
final columnValues = (columns ?? row.table.managedColumns)
|
||||
.where((c) => c.columnName != 'id' && c.columnName != 'scopeId')
|
||||
final columnValues = (columns ?? row.table.managedColumns).crdtSyncableColumns
|
||||
.map((c) => ColumnValue(c, values[c.columnName]))
|
||||
.toList();
|
||||
|
||||
|
|
|
|||
|
|
@ -1,28 +1,10 @@
|
|||
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,
|
||||
});
|
||||
typedef _MergeNodes = ({
|
||||
Map<UuidValue, CrdtNode> nodesByUuid,
|
||||
Map<UuidValue, CrdtScopeNode> scopeNodesByUuid,
|
||||
});
|
||||
|
||||
/// Adds merge-specific behavior to [CrdtMutationRecorder].
|
||||
extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
||||
CrdtUniqueConflictResolver get _uniqueConflictResolver =>
|
||||
CrdtUniqueConflictResolver(this);
|
||||
|
||||
/// Locks the current user row so merges can serialize with other work.
|
||||
Future<void> lockCurrentUser(Transaction transaction) async {
|
||||
final user = _getEffectiveScope(transaction);
|
||||
final user = _context.effectiveScopeFor(transaction);
|
||||
// Use a row lock without fetching the record since the merge path only
|
||||
// needs serialization against concurrent work for the same user.
|
||||
await CrdtScope.db.lockRows(
|
||||
|
|
@ -42,10 +24,9 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
if (mergeSet.isEmpty) return;
|
||||
|
||||
final operations = mergeSet.causallyOrderedChanges;
|
||||
final shouldProjectForeignKeys = _mergeOperationsMayAffectForeignKeys(
|
||||
operations,
|
||||
);
|
||||
final currentUser = _getEffectiveScope(transaction);
|
||||
final shouldProjectForeignKeys = _foreignKeyProjector
|
||||
.mergeOperationsMayAffectForeignKeys(operations);
|
||||
final currentUser = _context.effectiveScopeFor(transaction);
|
||||
final remoteNodes = await _findOrCreateNodesForMerge(
|
||||
currentUser.id!,
|
||||
{for (final change in operations) change.uuidNodeId},
|
||||
|
|
@ -55,7 +36,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
final context = await _loadMergeContext(mergeSet, transaction);
|
||||
|
||||
for (final operation in operations) {
|
||||
if (!_isCrdtTrackedTableName(operation.tableName)) {
|
||||
if (!_context.isCrdtTrackedTableName(operation.tableName)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
|
|
@ -85,7 +66,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
}
|
||||
|
||||
if (shouldProjectForeignKeys) {
|
||||
await _projectForeignKeys(transaction);
|
||||
await _foreignKeyProjector.project(transaction);
|
||||
}
|
||||
await _updateHlcFromIncomingOperations(
|
||||
operations,
|
||||
|
|
@ -96,7 +77,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
|
||||
Future<void> _updateHlcFromIncomingOperations(
|
||||
List<CrdtMergeChange> operations,
|
||||
_MergeNodes remoteNodes,
|
||||
MergeNodes remoteNodes,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final maxIncomingHlc = operations.fold<Hlc?>(
|
||||
|
|
@ -106,7 +87,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
final scopeNodesToUpdate = <CrdtScopeNode>[];
|
||||
|
||||
if (maxIncomingHlc != null) {
|
||||
final hlcManager = _getHlcManager(transaction);
|
||||
final hlcManager = _context.hlcManagerFor(transaction);
|
||||
if (maxIncomingHlc.nodeId == hlcManager.uuidNodeId) {
|
||||
if (maxIncomingHlc > hlcManager.lastHlc) {
|
||||
hlcManager.lastHlc = maxIncomingHlc;
|
||||
|
|
@ -115,7 +96,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
hlcManager.merge(maxIncomingHlc);
|
||||
}
|
||||
|
||||
await _persistCurrentNodeHlc(hlcManager, transaction);
|
||||
await _context.persistCurrentNodeHlc(hlcManager, transaction);
|
||||
}
|
||||
|
||||
final maxIncomingHlcByNode = <UuidValue, Hlc>{};
|
||||
|
|
@ -150,7 +131,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
}
|
||||
}
|
||||
|
||||
Future<_MergeNodes> _findOrCreateNodesForMerge(
|
||||
Future<MergeNodes> _findOrCreateNodesForMerge(
|
||||
int userId,
|
||||
Set<UuidValue> nodeIds,
|
||||
Transaction transaction,
|
||||
|
|
@ -164,12 +145,12 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
|
||||
final nodesByUuid = {
|
||||
for (final uuidNodeId in nodeIds)
|
||||
uuidNodeId: await _findOrCreateNode(uuidNodeId, transaction),
|
||||
uuidNodeId: await _context.findOrCreateNode(uuidNodeId, transaction),
|
||||
};
|
||||
|
||||
final scopeNodesByUuid = {
|
||||
for (final entry in nodesByUuid.entries)
|
||||
entry.key: await _findOrCreateScopeNode(
|
||||
entry.key: await _context.findOrCreateScopeNode(
|
||||
userId,
|
||||
entry.value.id!,
|
||||
transaction,
|
||||
|
|
@ -182,31 +163,31 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
);
|
||||
}
|
||||
|
||||
Future<_MergeContext> _loadMergeContext(
|
||||
Future<MergeContext> _loadMergeContext(
|
||||
CrdtMergeSet mergeSet,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final (:rowIdsByTable, :columnNamesByTable) = mergeSet.collectMetadataLookup(
|
||||
columnNamesByTableName: _syncedTableColumnNamesForMerge,
|
||||
columnNamesByTableName: _context.syncedTableColumnNamesForMerge,
|
||||
);
|
||||
|
||||
final rows = <_MergeRowKey, CrdtDataRow>{};
|
||||
final fields = <_MergeFieldKey, CrdtDataField>{};
|
||||
final rows = <MergeRowKey, CrdtDataRow>{};
|
||||
final fields = <MergeFieldKey, CrdtDataField>{};
|
||||
final incomingFieldHlcs = {
|
||||
for (final update in mergeSet.whereType<CrdtMergeUpdate>())
|
||||
(update.tableName, update.uuidRowId, update.columnName): update.hlc,
|
||||
};
|
||||
final tombstones = <_MergeRowKey, CrdtDataDeleted>{};
|
||||
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>{};
|
||||
final domainOwners = <MergeRowKey, DomainRowOwner>{};
|
||||
|
||||
for (final MapEntry(key: tableName, value: rowIds) in rowIdsByTable.entries) {
|
||||
final loadedRows = await _findCrdtRows(
|
||||
final loadedRows = await _context.findCrdtRows(
|
||||
tableName,
|
||||
rowIds,
|
||||
transaction,
|
||||
|
|
@ -225,7 +206,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
|
||||
final trackedRowIds = loadedRows.map((row) => row.uuidRowId).toSet();
|
||||
if (trackedRowIds.isNotEmpty) {
|
||||
final ownerValues = await _readDomainColumnValues(
|
||||
final ownerValues = await _context.readDomainColumnValues(
|
||||
tableName,
|
||||
trackedRowIds,
|
||||
['scopeId'],
|
||||
|
|
@ -250,7 +231,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
);
|
||||
}
|
||||
|
||||
final columnIdsByName = _schemaColumnIds(_schema, tableName, columnNames);
|
||||
final columnIdsByName = _context.schemaColumnIds(tableName, columnNames);
|
||||
if (columnIdsByName.isEmpty) continue;
|
||||
|
||||
final loadedFields = await CrdtDataField.db.find(
|
||||
|
|
@ -282,7 +263,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
Future<void> _applyMergeInsert(
|
||||
CrdtMergeInsert insert,
|
||||
Map<UuidValue, CrdtNode> remoteNodes,
|
||||
_MergeContext context,
|
||||
MergeContext context,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final rowKey = (insert.tableName, insert.uuidRowId);
|
||||
|
|
@ -297,7 +278,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
final data = _sanitizeMergeRowData(insert.tableName, insert.databaseColumns);
|
||||
|
||||
if (currentRow == null) {
|
||||
final mergingScope = _getEffectiveScope(transaction);
|
||||
final mergingScope = _context.effectiveScopeFor(transaction);
|
||||
final mergingScopeId = mergingScope.id!;
|
||||
try {
|
||||
// Insert-first inside a savepoint: the domain primary key resolves
|
||||
|
|
@ -309,10 +290,14 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
_db,
|
||||
transaction,
|
||||
(savepoint) async {
|
||||
final resolvedInsert = await _uniqueConflictResolver
|
||||
._resolveForIncomingInsert(insert, data, context, savepoint);
|
||||
final resolvedInsert = await _uniqueResolver.resolveForIncomingInsert(
|
||||
insert,
|
||||
data,
|
||||
context,
|
||||
savepoint,
|
||||
);
|
||||
|
||||
final safeInsert = await _safeIncomingForeignKeyData(
|
||||
final safeInsert = await _foreignKeyProjector.safeIncomingData(
|
||||
insert.tableName,
|
||||
insert.uuidRowId,
|
||||
resolvedInsert.data,
|
||||
|
|
@ -326,7 +311,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
safeInsert.data,
|
||||
savepoint,
|
||||
);
|
||||
await _recordForeignKeyInsertAttempts(
|
||||
await _foreignKeyProjector.recordInsertAttempts(
|
||||
insert.tableName,
|
||||
{insert.uuidRowId},
|
||||
savepoint,
|
||||
|
|
@ -368,7 +353,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
Future<void> _applyMergeUpdate(
|
||||
CrdtMergeUpdate update,
|
||||
Map<UuidValue, CrdtNode> remoteNodes,
|
||||
_MergeContext context,
|
||||
MergeContext context,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final rowKey = (update.tableName, update.uuidRowId);
|
||||
|
|
@ -401,25 +386,25 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
|
||||
if (!shouldApply) return;
|
||||
|
||||
final resolvedUpdates = await _uniqueConflictResolver._resolveForIncomingUpdate(
|
||||
final resolvedUpdates = await _uniqueResolver.resolveForIncomingUpdate(
|
||||
update,
|
||||
row,
|
||||
context,
|
||||
transaction,
|
||||
);
|
||||
final safeUpdate = await _safeIncomingForeignKeyData(
|
||||
final safeUpdate = await _foreignKeyProjector.safeIncomingData(
|
||||
update.tableName,
|
||||
update.uuidRowId,
|
||||
resolvedUpdates,
|
||||
transaction,
|
||||
);
|
||||
await _updateDomainRows(
|
||||
await _context.updateDomainRows(
|
||||
update.tableName,
|
||||
{update.uuidRowId},
|
||||
safeUpdate.data,
|
||||
transaction,
|
||||
);
|
||||
await _recordForeignKeyAttemptsForRows(
|
||||
await _foreignKeyProjector.recordAttemptsForRows(
|
||||
update.tableName,
|
||||
{update.uuidRowId},
|
||||
resolvedUpdates.keys.toSet(),
|
||||
|
|
@ -431,7 +416,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
Future<void> _applyMergeDelete(
|
||||
CrdtMergeDelete delete,
|
||||
Map<UuidValue, CrdtNode> remoteNodes,
|
||||
_MergeContext context,
|
||||
MergeContext context,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final rowKey = (delete.tableName, delete.uuidRowId);
|
||||
|
|
@ -469,7 +454,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
);
|
||||
|
||||
if (delete.reason.isSynced) {
|
||||
await _applyRowVisibilityFromUserTombstone(
|
||||
await _context.applyRowVisibilityFromUserTombstone(
|
||||
row,
|
||||
context.tombstones[rowKey]!,
|
||||
transaction,
|
||||
|
|
@ -486,11 +471,11 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
Transaction transaction,
|
||||
) async {
|
||||
if (currentRow == null) {
|
||||
final (tableId, _) = _schema[tableName]!;
|
||||
final (tableId, _) = _context.schema[tableName]!;
|
||||
final insertedRow = await CrdtDataRow.db.insertRow(
|
||||
_session,
|
||||
CrdtDataRow(
|
||||
scopeId: _getHlcManager(transaction).normalizedScopeId,
|
||||
scopeId: _context.hlcManagerFor(transaction).normalizedScopeId,
|
||||
tblId: tableId,
|
||||
uuidRowId: rowId,
|
||||
nodeId: remoteNode.id!,
|
||||
|
|
@ -633,8 +618,8 @@ 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);
|
||||
final mergingScopeId = _context.hlcManagerFor(transaction).normalizedScopeId;
|
||||
final scopedRow = patchedRow.copyWithScopeId(mergingScopeId);
|
||||
try {
|
||||
// Insert first: the domain primary key resolves creation races
|
||||
// atomically. The inner savepoint keeps a violation from poisoning the
|
||||
|
|
@ -679,7 +664,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
CrdtNode remoteNode,
|
||||
Hlc incomingHlc,
|
||||
Map<String, Object?> data,
|
||||
_MergeContext context,
|
||||
MergeContext context,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
if (!await _domainRowOwnedByEffectiveScope(
|
||||
|
|
@ -695,7 +680,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
return;
|
||||
}
|
||||
|
||||
final (_, columnsByName) = _schema[insert.tableName]!;
|
||||
final (_, columnsByName) = _context.schema[insert.tableName]!;
|
||||
final updatedValues = <String, Object?>{};
|
||||
for (final MapEntry(key: columnName, value: schemaColumn)
|
||||
in columnsByName.entries) {
|
||||
|
|
@ -719,7 +704,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
|
||||
var updatesToApply = updatedValues;
|
||||
if (updatesToApply.isNotEmpty) {
|
||||
updatesToApply = await _uniqueConflictResolver._resolveForIncomingUpdates(
|
||||
updatesToApply = await _uniqueResolver.resolveForIncomingUpdates(
|
||||
tableName: insert.tableName,
|
||||
row: currentRow,
|
||||
updates: updatesToApply,
|
||||
|
|
@ -729,19 +714,19 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
}
|
||||
|
||||
if (updatesToApply.isNotEmpty) {
|
||||
final safeUpdate = await _safeIncomingForeignKeyData(
|
||||
final safeUpdate = await _foreignKeyProjector.safeIncomingData(
|
||||
insert.tableName,
|
||||
insert.uuidRowId,
|
||||
updatesToApply,
|
||||
transaction,
|
||||
);
|
||||
await _updateDomainRows(
|
||||
await _context.updateDomainRows(
|
||||
insert.tableName,
|
||||
{insert.uuidRowId},
|
||||
safeUpdate.data,
|
||||
transaction,
|
||||
);
|
||||
await _recordForeignKeyAttemptsForRows(
|
||||
await _foreignKeyProjector.recordAttemptsForRows(
|
||||
insert.tableName,
|
||||
{insert.uuidRowId},
|
||||
updatesToApply.keys.toSet(),
|
||||
|
|
@ -763,7 +748,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
currentTombstone,
|
||||
transaction,
|
||||
);
|
||||
await _applyRowVisibilityFromUserTombstone(
|
||||
await _context.applyRowVisibilityFromUserTombstone(
|
||||
currentRow,
|
||||
context.tombstones[rowKey]!,
|
||||
transaction,
|
||||
|
|
@ -778,12 +763,12 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
required CrdtDataRow row,
|
||||
required CrdtNode remoteNode,
|
||||
required Hlc incomingHlc,
|
||||
required Map<_MergeFieldKey, CrdtDataField> fields,
|
||||
required Map<MergeFieldKey, CrdtDataField> fields,
|
||||
required Transaction transaction,
|
||||
CrdtSchemaColumn? schemaColumn,
|
||||
}) async {
|
||||
final resolvedSchemaColumn =
|
||||
schemaColumn ?? _schemaColumn(_schema, tableName, columnName);
|
||||
schemaColumn ?? _context.schemaColumn(tableName, columnName);
|
||||
if (resolvedSchemaColumn == null) return false;
|
||||
|
||||
final fieldKey = (tableName, rowId, columnName);
|
||||
|
|
@ -826,7 +811,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
String tableName,
|
||||
Map<String, Object?> data,
|
||||
) {
|
||||
final columns = _context._columnsByTableAndName[tableName];
|
||||
final columns = _context.columnsByTableAndName[tableName];
|
||||
if (columns == null) return {};
|
||||
|
||||
return {
|
||||
|
|
@ -835,17 +820,17 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
columnName != 'scopeId' &&
|
||||
columns.containsKey(columnName))
|
||||
columnName: columns[columnName]!.columnType == ColumnType.uuid
|
||||
? _uuidValueFromDatabase(value)
|
||||
? value.toUuidValue()
|
||||
: value,
|
||||
};
|
||||
}
|
||||
|
||||
Future<({bool exists, int? scopeId})> _readDomainRowOwner(
|
||||
Future<DomainRowOwner> _readDomainRowOwner(
|
||||
String tableName,
|
||||
UuidValue rowId,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final values = await _readDomainColumnValues(
|
||||
final values = await _context.readDomainColumnValues(
|
||||
tableName,
|
||||
{rowId},
|
||||
['scopeId'],
|
||||
|
|
@ -859,8 +844,8 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
/// 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,
|
||||
Future<DomainRowOwner> _resolveDomainOwner(
|
||||
MergeContext context,
|
||||
String tableName,
|
||||
UuidValue rowId,
|
||||
Transaction transaction,
|
||||
|
|
@ -877,14 +862,14 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
required CrdtSyncViolationOperation operation,
|
||||
required String tableName,
|
||||
required UuidValue rowId,
|
||||
required _MergeContext context,
|
||||
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);
|
||||
final mergingScope = _context.effectiveScopeFor(transaction);
|
||||
if (owner.exists && owner.scopeId == mergingScope.id) return true;
|
||||
if (!owner.exists) return false;
|
||||
|
||||
|
|
@ -945,10 +930,6 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
);
|
||||
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
|
||||
|
|
@ -964,10 +945,8 @@ extension on DatabaseSerializationManager {
|
|||
/// Patches a [TableRow] with the given data.
|
||||
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' &&
|
||||
column.columnName != 'scopeId' &&
|
||||
data.containsKey(column.columnName)) {
|
||||
for (final column in row.table.crdtSyncableColumns) {
|
||||
if (data.containsKey(column.columnName)) {
|
||||
rowJson[column.fieldName] = data[column.columnName];
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,108 @@
|
|||
import 'package:meta/meta.dart';
|
||||
import 'package:serverpod_database/serverpod_database.dart';
|
||||
import 'package:serverpod_serialization/serverpod_serialization.dart';
|
||||
|
||||
/// Extensions on [String] used for SQL construction.
|
||||
@internal
|
||||
extension SqlStringExtension on String {
|
||||
/// Escapes this SQL identifier so it can be safely wrapped in double quotes.
|
||||
String escapeIdentifier() => replaceAll('"', '""');
|
||||
}
|
||||
|
||||
/// Extensions on [Object?] used for SQL construction.
|
||||
@internal
|
||||
extension SqlObjectExtension on Object? {
|
||||
/// Encodes this value as a SQL literal.
|
||||
String sqlLiteral() => ValueEncoder.instance.convert(this);
|
||||
}
|
||||
|
||||
/// Extensions on [Iterable] used for SQL construction.
|
||||
@internal
|
||||
extension SqlIterableExtension on Iterable<Object?> {
|
||||
/// Encodes these values as a comma-separated list of SQL literals.
|
||||
String sqlLiteralList() => map((value) => value.sqlLiteral()).join(', ');
|
||||
}
|
||||
|
||||
/// SQL predicate matching rows where [columnName] equals [value].
|
||||
@internal
|
||||
String domainColumnPredicate(
|
||||
String columnName,
|
||||
Object? value, {
|
||||
String alias = 'd',
|
||||
}) => '$alias."${columnName.escapeIdentifier()}" = ${value.sqlLiteral()}';
|
||||
|
||||
/// SQL predicate matching rows where [columnName] differs from [value].
|
||||
@internal
|
||||
String domainColumnNotPredicate(
|
||||
String columnName,
|
||||
Object? value, {
|
||||
String alias = 'd',
|
||||
}) => '$alias."${columnName.escapeIdentifier()}" <> ${value.sqlLiteral()}';
|
||||
|
||||
/// Extensions on [Object?] used for UUID conversion.
|
||||
@internal
|
||||
extension UuidValueExtension on Object? {
|
||||
/// Converts this raw database value into a [UuidValue], when present.
|
||||
UuidValue? toUuidValue() {
|
||||
final value = this;
|
||||
if (value == null) return null;
|
||||
if (value is UuidValue) return value;
|
||||
if (value is String) return UuidValue.withValidation(value);
|
||||
return UuidValueJsonExtension.fromJson(value);
|
||||
}
|
||||
}
|
||||
|
||||
/// Extensions on [UuidValue?] used for UUID comparison.
|
||||
@internal
|
||||
extension UuidValueComparisonExtension on UuidValue? {
|
||||
/// Whether this UUID and [other] represent the same UUID (or are both null).
|
||||
bool sameUuidValue(UuidValue? other) => this?.uuid == other?.uuid;
|
||||
}
|
||||
|
||||
/// Extensions on domain [TableRow]s used by the CRDT recorder.
|
||||
@internal
|
||||
extension CrdtTableRowExtension<T extends TableRow> on T {
|
||||
/// Returns a copy of this row with [scopeId] set.
|
||||
/// Must use dynamic cast because the copyWith method is generated only.
|
||||
T copyWithScopeId(int scopeId) => (this as dynamic).copyWith(scopeId: scopeId) as T;
|
||||
}
|
||||
|
||||
/// Helpers over lists of domain [TableRow]s used by the CRDT recorder.
|
||||
@internal
|
||||
extension TableRowListExtension on List<TableRow> {
|
||||
/// The UUID primary keys of all rows in the list.
|
||||
Set<UuidValue> get uuidRowIds {
|
||||
if (isEmpty) return {};
|
||||
if (first.id == null) throw StateError('Row IDs must be non-null.');
|
||||
if (first.id is! UuidValue) throw StateError('Row IDs must be UuidValue.');
|
||||
return {for (final row in this) row.id as UuidValue};
|
||||
}
|
||||
}
|
||||
|
||||
/// Extensions on [Table] used by the CRDT recorder.
|
||||
@internal
|
||||
extension CrdtTableExtension on Table {
|
||||
/// The scope ownership column managed by the CRDT layer, if present.
|
||||
ColumnInt? get crdtScopeIdColumn {
|
||||
for (final column in columns) {
|
||||
if (column.columnName == 'scopeId' && column is ColumnInt) {
|
||||
return column;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/// The columns that are part of CRDT sync, excluding the primary key and the
|
||||
/// CRDT-managed scopeId column.
|
||||
Iterable<Column> get crdtSyncableColumns => managedColumns.crdtSyncableColumns;
|
||||
}
|
||||
|
||||
/// Extensions on iterables of [Column] used by the CRDT recorder.
|
||||
@internal
|
||||
extension CrdtColumnIterableExtension on Iterable<Column> {
|
||||
/// The columns that are part of CRDT sync, excluding the primary key and the
|
||||
/// CRDT-managed scopeId column.
|
||||
Iterable<Column> get crdtSyncableColumns => where(
|
||||
(column) => column.columnName != 'id' && column.columnName != 'scopeId',
|
||||
);
|
||||
}
|
||||
|
|
@ -0,0 +1,147 @@
|
|||
import 'package:meta/meta.dart';
|
||||
import 'package:serverpod_database/serverpod_database.dart';
|
||||
|
||||
import '../recorder.dart';
|
||||
|
||||
/// A foreign key pointing at a parent table, seen from the parent side.
|
||||
@internal
|
||||
typedef ReferencingForeignKey = ({
|
||||
String childTableName,
|
||||
String childColumn,
|
||||
String parentColumn,
|
||||
ForeignKeyAction action,
|
||||
});
|
||||
|
||||
/// A foreign key between two CRDT-tracked tables, with repair metadata.
|
||||
@internal
|
||||
typedef ForeignKeyEdge = ({
|
||||
String childTableName,
|
||||
String childColumn,
|
||||
String parentTableName,
|
||||
String parentColumn,
|
||||
ForeignKeyAction action,
|
||||
bool childNullable,
|
||||
Object? defaultValue,
|
||||
});
|
||||
|
||||
/// Identifies a foreign key value: `(tableName, columnName, value)`.
|
||||
@internal
|
||||
typedef ForeignKeyValueKey = (String tableName, String columnName, String value);
|
||||
|
||||
/// Foreign key relationships derived from the schema, indexed for lookups.
|
||||
@internal
|
||||
class CrdtForeignKeyGraph {
|
||||
/// Creates a [CrdtForeignKeyGraph] over the tables known to the context.
|
||||
CrdtForeignKeyGraph(this._context);
|
||||
|
||||
final CrdtDatabaseContext _context;
|
||||
|
||||
/// Foreign keys grouped by the table they reference.
|
||||
///
|
||||
/// Covers all known tables, not only CRDT-tracked ones, and rejects
|
||||
/// composite foreign keys which soft deletes cannot handle.
|
||||
late final Map<String, List<ReferencingForeignKey>> referencingKeysByParentTable =
|
||||
() {
|
||||
final result = <String, List<ReferencingForeignKey>>{};
|
||||
for (final table in _context.tableDefinitionsByName.values) {
|
||||
final childTableName = table.name;
|
||||
// Untracked children have no CRDT schema rows and cannot be
|
||||
// soft-deleted; their physical rows keep referencing the (physically
|
||||
// preserved) soft-deleted parent, so no delete action applies.
|
||||
// Interim behavior: docs/sync-non-sync-relations.md forbids
|
||||
// non-synced -> synced relations outright, but its initialize()
|
||||
// validation is not implemented yet.
|
||||
if (!_context.isCrdtTrackedTableName(childTableName)) continue;
|
||||
for (final foreignKey in table.foreignKeys) {
|
||||
if (foreignKey.columns.length != 1 ||
|
||||
foreignKey.referenceColumns.length != 1) {
|
||||
throw StateError(
|
||||
'Composite foreign keys are not supported for CRDT soft deletes.',
|
||||
);
|
||||
}
|
||||
|
||||
result.putIfAbsent(foreignKey.referenceTable, () => []).add((
|
||||
childTableName: childTableName,
|
||||
childColumn: foreignKey.columns.single,
|
||||
parentColumn: foreignKey.referenceColumns.single,
|
||||
action: foreignKey.onDelete ?? ForeignKeyAction.noAction,
|
||||
));
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}();
|
||||
|
||||
/// Foreign key edges between CRDT-tracked tables.
|
||||
late final List<ForeignKeyEdge> edges = [
|
||||
for (final table in _context.tableDefinitionsByName.values)
|
||||
if (_context.isCrdtTrackedTableName(table.name))
|
||||
for (final foreignKey in table.foreignKeys)
|
||||
if (foreignKey.columns.length == 1 &&
|
||||
foreignKey.referenceColumns.length == 1 &&
|
||||
_context.isCrdtTrackedTableName(foreignKey.referenceTable))
|
||||
(
|
||||
childTableName: table.name,
|
||||
childColumn: foreignKey.columns.single,
|
||||
parentTableName: foreignKey.referenceTable,
|
||||
parentColumn: foreignKey.referenceColumns.single,
|
||||
action: foreignKey.onDelete ?? ForeignKeyAction.noAction,
|
||||
childNullable: _context
|
||||
.columnsByTableAndName[table.name]![foreignKey.columns.single]!
|
||||
.isNullable,
|
||||
defaultValue: _context.defaultValueForColumn(
|
||||
table.name,
|
||||
foreignKey.columns.single,
|
||||
),
|
||||
),
|
||||
];
|
||||
|
||||
/// [edges] grouped by their parent table name.
|
||||
late final Map<String, List<ForeignKeyEdge>> edgesByParentTable = {
|
||||
for (final tableName in _context.tableDefinitionsByName.keys)
|
||||
tableName: [
|
||||
for (final edge in edges)
|
||||
if (edge.parentTableName == tableName) edge,
|
||||
],
|
||||
};
|
||||
|
||||
/// [edges] grouped by their child table name.
|
||||
late final Map<String, List<ForeignKeyEdge>> edgesByChildTable = {
|
||||
for (final tableName in _context.tableDefinitionsByName.keys)
|
||||
tableName: [
|
||||
for (final edge in edges)
|
||||
if (edge.childTableName == tableName) edge,
|
||||
],
|
||||
};
|
||||
|
||||
/// Whether changing the given columns may affect foreign key projection.
|
||||
///
|
||||
/// Pass null [columnNames] when all columns may change (inserts/deletes).
|
||||
bool columnsMayAffectForeignKeys(
|
||||
String tableName,
|
||||
Set<String>? columnNames,
|
||||
) {
|
||||
final childEdges = edgesByChildTable[tableName] ?? const <ForeignKeyEdge>[];
|
||||
final parentEdges = edgesByParentTable[tableName] ?? const <ForeignKeyEdge>[];
|
||||
if (childEdges.isEmpty && parentEdges.isEmpty) return false;
|
||||
if (columnNames == null) return true;
|
||||
|
||||
return childEdges.any((edge) => columnNames.contains(edge.childColumn)) ||
|
||||
parentEdges.any(
|
||||
(edge) =>
|
||||
edge.parentColumn != 'id' && columnNames.contains(edge.parentColumn),
|
||||
);
|
||||
}
|
||||
|
||||
/// The child column names of foreign keys outgoing from [tableName].
|
||||
///
|
||||
/// When [columnNames] is provided, only edges whose child column is in the set
|
||||
/// are included; otherwise all outgoing FK columns are returned.
|
||||
Set<String> foreignKeyColumnsFor(
|
||||
String tableName, {
|
||||
Set<String>? columnNames,
|
||||
}) => {
|
||||
for (final edge in edgesByChildTable[tableName] ?? const <ForeignKeyEdge>[])
|
||||
if (columnNames == null || columnNames.contains(edge.childColumn))
|
||||
edge.childColumn,
|
||||
};
|
||||
}
|
||||
File diff suppressed because it is too large
Load diff
|
|
@ -0,0 +1,645 @@
|
|||
import 'package:meta/meta.dart';
|
||||
import 'package:serverpod_database/serverpod_database.dart';
|
||||
import 'package:serverpod_serialization/serverpod_serialization.dart';
|
||||
|
||||
import '../../crdt/extensions.dart';
|
||||
import '../../generated/protocol.dart';
|
||||
import '../../managers/hlc.dart';
|
||||
import '../../managers/scope.dart';
|
||||
import '../database.dart';
|
||||
import '../recorder.dart';
|
||||
import '../session.dart';
|
||||
import '../unique_index_utils.dart';
|
||||
import 'database_helpers.dart';
|
||||
import 'types.dart';
|
||||
|
||||
/// Whether a foreign key target row is absent, visible, or hidden.
|
||||
@internal
|
||||
enum ForeignKeyTargetPresence {
|
||||
absent,
|
||||
visible,
|
||||
hidden,
|
||||
}
|
||||
|
||||
/// Shared state and database access for the CRDT recorder and its helpers.
|
||||
///
|
||||
/// Owns per-recorder scope/HLC management and the low-level domain and CRDT
|
||||
/// metadata queries shared by the mutation recorder, the unique conflict
|
||||
/// resolver, and the foreign key projector.
|
||||
@internal
|
||||
class CrdtRecorderContext {
|
||||
/// Creates a [CrdtRecorderContext] instance.
|
||||
CrdtRecorderContext(
|
||||
this.database, {
|
||||
required this.databaseContext,
|
||||
required this.persistentUserId,
|
||||
});
|
||||
|
||||
/// The underlying user database (not the CRDT proxy).
|
||||
final Database database;
|
||||
|
||||
/// Process-level CRDT metadata shared across recorders.
|
||||
final CrdtDatabaseContext databaseContext;
|
||||
|
||||
/// The user ID to use for all CRDT operations. This should only be used for
|
||||
/// databases operating on the client side, where all data is for the same user.
|
||||
/// Otherwise, the user ID must be passed through the transaction.
|
||||
final UuidValue? persistentUserId;
|
||||
|
||||
/// A plain session over [database] for metadata queries.
|
||||
late final DatabaseSession databaseSession = database.session;
|
||||
|
||||
/// Manages [CrdtScope] rows and their cache.
|
||||
late final CrdtScopeManager scopeManager = CrdtScopeManager(databaseSession);
|
||||
|
||||
final Map<UuidValue, HlcManager> _hlcManagers = {};
|
||||
|
||||
/// Clears cached [HlcManager] instances for this recorder session.
|
||||
void clearHlcManagers() => _hlcManagers.clear();
|
||||
|
||||
/// CRDT schema ids by table name: `tableName -> (tableId, columnsByName)`.
|
||||
Map<String, (int, Map<String, CrdtSchemaColumn>)> get schema =>
|
||||
databaseContext.schema;
|
||||
|
||||
/// The list of tables to sync with CRDT.
|
||||
List<Table> get syncTables => databaseContext.syncTables;
|
||||
|
||||
/// Returns the local [CrdtSchemaTable] id for [tableName], or null when the
|
||||
/// table is not registered for CRDT synchronization.
|
||||
int? tableIdForName(String tableName) => databaseContext.tableIdForName(tableName);
|
||||
|
||||
/// Whether the given table name is tracked by CRDT.
|
||||
bool isCrdtTrackedTableName(String tableName) =>
|
||||
databaseContext.isCrdtTrackedTableName(tableName);
|
||||
|
||||
/// The CRDT schema column for the given table and column names, if any.
|
||||
CrdtSchemaColumn? schemaColumn(String tableName, String columnName) =>
|
||||
databaseContext.schemaColumn(tableName, columnName);
|
||||
|
||||
/// The CRDT schema column ids for the given table and column names.
|
||||
///
|
||||
/// Columns without a CRDT schema entry are omitted from the result.
|
||||
Map<String, int> schemaColumnIds(
|
||||
String tableName,
|
||||
Iterable<String> columnNames,
|
||||
) => databaseContext.schemaColumnIds(tableName, columnNames);
|
||||
|
||||
/// Serverpod table definitions by table name.
|
||||
Map<String, TableDefinition> get tableDefinitionsByName =>
|
||||
databaseContext.tableDefinitionsByName;
|
||||
|
||||
/// Column definitions by table name and column name.
|
||||
Map<String, Map<String, ColumnDefinition>> get columnsByTableAndName =>
|
||||
databaseContext.columnsByTableAndName;
|
||||
|
||||
/// Synced tables by table name.
|
||||
Map<String, Table> get syncTableByName => databaseContext.syncTableByName;
|
||||
|
||||
/// Column names of every synced table, used to scope merge metadata lookups.
|
||||
Map<String, Set<String>> get syncedTableColumnNamesForMerge =>
|
||||
databaseContext.syncedTableColumnNamesForMerge;
|
||||
|
||||
/// The default value declared for the given column, if usable.
|
||||
Object? defaultValueForColumn(String tableName, String columnName) =>
|
||||
databaseContext.defaultValueForColumn(tableName, columnName);
|
||||
|
||||
Future<List<CrdtDataRow>> findCrdtRows(
|
||||
String tableName,
|
||||
Set<UuidValue> rowIds,
|
||||
Transaction transaction, {
|
||||
CrdtDataRowInclude? include,
|
||||
}) {
|
||||
final (tableId, _) = schema[tableName]!;
|
||||
final scopeId = hlcManagerFor(transaction).normalizedScopeId;
|
||||
return CrdtDataRow.db.find(
|
||||
databaseSession,
|
||||
where: (t) =>
|
||||
t.scopeId.equals(scopeId) &
|
||||
t.tblId.equals(tableId) &
|
||||
t.uuidRowId.inSet(rowIds),
|
||||
include: include,
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
Future<List<CrdtDataRow>> findRequiredCrdtRows(
|
||||
String tableName,
|
||||
Set<UuidValue> rowIds,
|
||||
String operation,
|
||||
Transaction transaction, {
|
||||
bool includeDeleted = false,
|
||||
}) async {
|
||||
final crdtDataRows = await findCrdtRows(
|
||||
tableName,
|
||||
rowIds,
|
||||
transaction,
|
||||
include: includeDeleted
|
||||
? CrdtDataRow.include(deleted: CrdtDataDeleted.include())
|
||||
: null,
|
||||
);
|
||||
if (crdtDataRows.length != rowIds.length) {
|
||||
final found = crdtDataRows.map((e) => e.uuidRowId).toSet();
|
||||
final missing = rowIds.difference(found);
|
||||
throw StateError(
|
||||
'Missing CRDT rows for ${missing.length} $operation domain rows:\n'
|
||||
'${missing.map((id) => ' - $id').join('\n')}',
|
||||
);
|
||||
}
|
||||
return crdtDataRows;
|
||||
}
|
||||
|
||||
Future<void> upsertCrdtFieldsForRows(
|
||||
String tableName,
|
||||
List<CrdtDataRow> crdtDataRows,
|
||||
List<CrdtSchemaColumn> schemaColumns,
|
||||
Transaction transaction, {
|
||||
Set<MergeFieldKey> skippedFields = const {},
|
||||
}) async {
|
||||
if (crdtDataRows.isEmpty || schemaColumns.isEmpty) return;
|
||||
|
||||
final rowPks = crdtDataRows.map((r) => r.id!).toSet();
|
||||
final columnPks = schemaColumns.map((c) => c.id!).toSet();
|
||||
final existingFields = await CrdtDataField.db.find(
|
||||
databaseSession,
|
||||
where: (t) => t.rowId.inSet(rowPks) & t.columnId.inSet(columnPks),
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
final fieldByRowAndColumn = {
|
||||
for (final f in existingFields) (f.rowId, f.columnId): f,
|
||||
};
|
||||
|
||||
final hlcManager = hlcManagerFor(transaction);
|
||||
final toInsert = <CrdtDataField>[];
|
||||
final toUpdate = <CrdtDataField>[];
|
||||
|
||||
for (final row in crdtDataRows) {
|
||||
for (final schemaCol in schemaColumns) {
|
||||
if (skippedFields.contains((tableName, row.uuidRowId, schemaCol.name))) {
|
||||
continue;
|
||||
}
|
||||
|
||||
final hlc = hlcManager.increment();
|
||||
final existing = fieldByRowAndColumn[(row.id!, schemaCol.id!)];
|
||||
if (existing == null) {
|
||||
toInsert.add(
|
||||
CrdtDataField(
|
||||
rowId: row.id!,
|
||||
columnId: schemaCol.id!,
|
||||
nodeId: hlcManager.normalizedNodeId,
|
||||
hlcDatetime: hlc.datetime,
|
||||
hlcCounter: hlc.counter,
|
||||
),
|
||||
);
|
||||
} else {
|
||||
toUpdate.add(
|
||||
existing.copyWith(
|
||||
nodeId: hlcManager.normalizedNodeId,
|
||||
hlcDatetime: hlc.datetime,
|
||||
hlcCounter: hlc.counter,
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (toInsert.isNotEmpty) {
|
||||
await CrdtDataField.db.insert(
|
||||
databaseSession,
|
||||
toInsert,
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
if (toUpdate.isNotEmpty) {
|
||||
await CrdtDataField.db.update(
|
||||
databaseSession,
|
||||
toUpdate,
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> recordFieldsUpdatedByTable(
|
||||
String tableName,
|
||||
Set<UuidValue> rowIds,
|
||||
List<String> columnNames,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
if (rowIds.isEmpty || columnNames.isEmpty) return;
|
||||
if (!isCrdtTrackedTableName(tableName)) return;
|
||||
|
||||
final crdtDataRows = await findCrdtRows(tableName, rowIds, transaction);
|
||||
if (crdtDataRows.isEmpty) return;
|
||||
|
||||
final (_, colMap) = schema[tableName]!;
|
||||
final schemaColumns = [
|
||||
for (final columnName in columnNames)
|
||||
if (colMap[columnName] != null) colMap[columnName]!,
|
||||
];
|
||||
if (schemaColumns.isEmpty) return;
|
||||
|
||||
await upsertCrdtFieldsForRows(
|
||||
tableName,
|
||||
crdtDataRows,
|
||||
schemaColumns,
|
||||
transaction,
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> markCrdtRowsDeleted(
|
||||
List<CrdtDataRow> crdtDataRows,
|
||||
// Preserved verbatim from the recorder; positional bool matches tombstone API.
|
||||
// ignore: avoid_positional_boolean_parameters
|
||||
bool isDeleted,
|
||||
CrdtDataDeletedReason reason,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final existingTombs = await CrdtDataDeleted.db.find(
|
||||
databaseSession,
|
||||
where: (t) => t.rowId.inSet(crdtDataRows.map((r) => r.id!).toSet()),
|
||||
transaction: transaction,
|
||||
);
|
||||
final tombByCrdtRowPk = {for (final t in existingTombs) t.rowId: t};
|
||||
|
||||
final hlcManager = hlcManagerFor(transaction);
|
||||
final toInsert = <CrdtDataDeleted>[];
|
||||
final toUpdate = <CrdtDataDeleted>[];
|
||||
|
||||
for (final row in crdtDataRows) {
|
||||
final rowPk = row.id!;
|
||||
final hlc = hlcManager.increment();
|
||||
final existing = tombByCrdtRowPk[rowPk];
|
||||
final clFlag = _nextClFlag(existing, isDeleted);
|
||||
|
||||
if (existing == null) {
|
||||
toInsert.add(
|
||||
CrdtDataDeleted(
|
||||
rowId: rowPk,
|
||||
nodeId: hlcManager.normalizedNodeId,
|
||||
hlcDatetime: hlc.datetime,
|
||||
hlcCounter: hlc.counter,
|
||||
clFlag: clFlag,
|
||||
reason: reason,
|
||||
),
|
||||
);
|
||||
} else {
|
||||
toUpdate.add(
|
||||
existing.copyWith(
|
||||
nodeId: hlcManager.normalizedNodeId,
|
||||
hlcDatetime: hlc.datetime,
|
||||
hlcCounter: hlc.counter,
|
||||
clFlag: clFlag,
|
||||
reason: reason,
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (toInsert.isNotEmpty) {
|
||||
await CrdtDataDeleted.db.insert(
|
||||
databaseSession,
|
||||
toInsert,
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
if (toUpdate.isNotEmpty) {
|
||||
await CrdtDataDeleted.db.update(
|
||||
databaseSession,
|
||||
toUpdate,
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
await applyRowVisibility(
|
||||
crdtDataRows,
|
||||
visibility: reason.toVisibility(isDeleted: isDeleted),
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> applyRowVisibility(
|
||||
List<CrdtDataRow> rows, {
|
||||
required CrdtDataRowVisibility visibility,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
final toUpdate = rows
|
||||
.where((row) => row.visibility != visibility)
|
||||
.map((row) => row.copyWith(visibility: visibility))
|
||||
.toList();
|
||||
if (toUpdate.isEmpty) return;
|
||||
|
||||
await CrdtDataRow.db.update(
|
||||
databaseSession,
|
||||
toUpdate,
|
||||
columns: (t) => [t.visibility],
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> applyRowVisibilityFromUserTombstone(
|
||||
CrdtDataRow row,
|
||||
CrdtDataDeleted tombstone,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
await applyRowVisibility(
|
||||
[row],
|
||||
visibility: tombstone.reason.toVisibility(isDeleted: tombstone.isDeleted),
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
Future<Set<UuidValue>> findVisibleReferencingRowIds({
|
||||
required String tableName,
|
||||
required String columnName,
|
||||
required Object? value,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
return findVisibleDomainRowIdsWhere(
|
||||
tableName: tableName,
|
||||
predicates: [domainColumnPredicate(columnName, value)],
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
Future<Set<UuidValue>> findVisibleDomainRowIdsWhere({
|
||||
required String tableName,
|
||||
required List<String> predicates,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
final (tableId, _) = schema[tableName]!;
|
||||
final scopeId = hlcManagerFor(transaction).normalizedScopeId;
|
||||
final result = await database.unsafeQuery(
|
||||
'''
|
||||
SELECT d."id"
|
||||
FROM "${tableName.escapeIdentifier()}" d
|
||||
LEFT JOIN "crdt_data_rows" r
|
||||
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)
|
||||
''',
|
||||
transaction: transaction,
|
||||
);
|
||||
return {
|
||||
for (final row in result) UuidValueJsonExtension.fromJson(row.first),
|
||||
};
|
||||
}
|
||||
|
||||
Future<ForeignKeyTargetPresence> lookupForeignKeyTargetPresence({
|
||||
required String parentTableName,
|
||||
required String parentColumn,
|
||||
required UuidValue value,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
final (tableId, _) = schema[parentTableName]!;
|
||||
final scopeId = hlcManagerFor(transaction).normalizedScopeId;
|
||||
final result = await database.unsafeQuery(
|
||||
'''
|
||||
SELECT CASE
|
||||
WHEN r."id" IS NULL OR r."visibility" <= $crdtRowLastVisibleVisibilityIndex THEN 1
|
||||
ELSE 0
|
||||
END AS visible
|
||||
FROM "${parentTableName.escapeIdentifier()}" d
|
||||
LEFT JOIN "crdt_data_rows" r
|
||||
ON r."scopeId" = $scopeId AND r."tblId" = $tableId AND r."uuidRowId" = d."id"
|
||||
WHERE (${domainColumnPredicate('scopeId', scopeId)})
|
||||
AND (${domainColumnPredicate(parentColumn, value)})
|
||||
LIMIT 1
|
||||
''',
|
||||
transaction: transaction,
|
||||
);
|
||||
if (result.isEmpty) return ForeignKeyTargetPresence.absent;
|
||||
return (result.first[0] as num) == 1
|
||||
? ForeignKeyTargetPresence.visible
|
||||
: ForeignKeyTargetPresence.hidden;
|
||||
}
|
||||
|
||||
Future<void> updateDomainRows(
|
||||
String tableName,
|
||||
Set<UuidValue> rowIds,
|
||||
Map<String, Object?> updates,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
if (rowIds.isEmpty || updates.isEmpty) return;
|
||||
|
||||
final assignments = updates.entries
|
||||
.map((e) => '"${e.key.escapeIdentifier()}" = ${e.value.sqlLiteral()}')
|
||||
.join(', ');
|
||||
|
||||
await database.unsafeExecute(
|
||||
'''
|
||||
UPDATE "${tableName.escapeIdentifier()}"
|
||||
SET $assignments
|
||||
WHERE "id" IN (${rowIds.sqlLiteralList()})
|
||||
''',
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
Future<Map<UuidValue, Map<String, Object?>>> readDomainColumnValues(
|
||||
String tableName,
|
||||
Set<UuidValue> rowIds,
|
||||
List<String> columnNames,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
if (rowIds.isEmpty || columnNames.isEmpty) return {};
|
||||
|
||||
final columns = [
|
||||
'"id"',
|
||||
for (final columnName in columnNames) '"${columnName.escapeIdentifier()}"',
|
||||
].join(', ');
|
||||
|
||||
final result = await database.unsafeQuery(
|
||||
'''
|
||||
SELECT $columns
|
||||
FROM "${tableName.escapeIdentifier()}"
|
||||
WHERE "id" IN (${rowIds.sqlLiteralList()})
|
||||
''',
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
return {
|
||||
for (final row in result)
|
||||
UuidValueJsonExtension.fromJson(row[0]): {
|
||||
for (final (index, columnName) in columnNames.indexed)
|
||||
columnName: row[index + 1],
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
Object? conflictFreeValue(
|
||||
UniqueColumnConflictRelease column,
|
||||
Object? value,
|
||||
String tableName,
|
||||
UuidValue conflictingId,
|
||||
String releaseSuffix,
|
||||
) {
|
||||
switch (column.kind) {
|
||||
case CrdtUniqueConflictReleaseKind.setNull:
|
||||
return null;
|
||||
case CrdtUniqueConflictReleaseKind.textSuffix:
|
||||
if (value is String) {
|
||||
return '${value}__${releaseSuffix}__${conflictingId.uuid}';
|
||||
}
|
||||
case CrdtUniqueConflictReleaseKind.syntheticUuid:
|
||||
if (value != null) {
|
||||
return const Uuid().v5obj(
|
||||
Namespace.oid.value,
|
||||
'$tableName.${column.columnName}:${value}__${releaseSuffix}__$conflictingId',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
throw StateError(
|
||||
'Unexpected value for $tableName.${column.columnName} while making '
|
||||
'tombstoned unique value conflict-free: ${value.runtimeType}.',
|
||||
);
|
||||
}
|
||||
|
||||
CrdtDataRow withNextHlc(CrdtDataRow row, HlcManager hlcManager) {
|
||||
final hlc = hlcManager.increment();
|
||||
return row.copyWith(
|
||||
nodeId: hlcManager.normalizedNodeId,
|
||||
hlcDatetime: hlc.datetime,
|
||||
hlcCounter: hlc.counter,
|
||||
);
|
||||
}
|
||||
|
||||
HlcManager hlcManagerFor(Transaction transaction) {
|
||||
final user = effectiveScopeFor(transaction);
|
||||
return _hlcManagers.putIfAbsent(
|
||||
user.uuidScopeId,
|
||||
() => HlcManager.forScope(user),
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> persistCurrentNodeHlc(
|
||||
HlcManager hlcManager,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
await CrdtNode.db.updateRow(
|
||||
databaseSession,
|
||||
hlcManager.getNode(),
|
||||
columns: (t) => [t.lastHlc],
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
Future<T> _findOrCreate<T>({
|
||||
required Future<T?> Function() find,
|
||||
required Future<void> Function() insert,
|
||||
required String label,
|
||||
}) async {
|
||||
final existing = await find();
|
||||
if (existing != null) return existing;
|
||||
|
||||
await insert();
|
||||
final created = await find();
|
||||
return created ?? (throw StateError('Could not create $label.'));
|
||||
}
|
||||
|
||||
Future<CrdtNode> findOrCreateNode(
|
||||
UuidValue uuidNodeId,
|
||||
Transaction transaction,
|
||||
) => _findOrCreate(
|
||||
find: () => CrdtNode.db.findFirstRow(
|
||||
databaseSession,
|
||||
where: (t) => t.uuidNodeId.equals(uuidNodeId),
|
||||
transaction: transaction,
|
||||
),
|
||||
insert: () => CrdtNode.db.insert(
|
||||
databaseSession,
|
||||
[CrdtNode(uuidNodeId: uuidNodeId)],
|
||||
transaction: transaction,
|
||||
ignoreConflicts: true,
|
||||
),
|
||||
label: 'CRDT node "$uuidNodeId"',
|
||||
);
|
||||
|
||||
Future<CrdtScopeNode> findOrCreateScopeNode(
|
||||
int scopeId,
|
||||
int nodeId,
|
||||
Transaction transaction,
|
||||
) => _findOrCreate(
|
||||
find: () => CrdtScopeNode.db.findFirstRow(
|
||||
databaseSession,
|
||||
where: (t) => t.scopeId.equals(scopeId) & t.nodeId.equals(nodeId),
|
||||
transaction: transaction,
|
||||
),
|
||||
insert: () => CrdtScopeNode.db.insert(
|
||||
databaseSession,
|
||||
[CrdtScopeNode(scopeId: scopeId, nodeId: nodeId)],
|
||||
transaction: transaction,
|
||||
ignoreConflicts: true,
|
||||
),
|
||||
label: 'CRDT scope-node row for scope $scopeId and node $nodeId',
|
||||
);
|
||||
|
||||
CrdtScope effectiveScopeFor(Transaction transaction) {
|
||||
final scope = scopeForTransaction[transaction];
|
||||
if (scope != null) return scope;
|
||||
final userId = persistentUserId;
|
||||
if (userId == null) {
|
||||
throw StateError('No user ID found for transaction or persistent user ID.');
|
||||
}
|
||||
return scopeManager.getCached(userId);
|
||||
}
|
||||
|
||||
Future<({int foreignKeyIndex, Object value})?> findInvalidForeignKeyReference({
|
||||
required String childTableName,
|
||||
required Set<UuidValue> childRowIds,
|
||||
required List<ForeignKeyDefinition> foreignKeys,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
if (foreignKeys.isEmpty) return null;
|
||||
|
||||
final scopeId = hlcManagerFor(transaction).normalizedScopeId;
|
||||
final escapedChildTable = childTableName.escapeIdentifier();
|
||||
final whereRowIds = childRowIds.sqlLiteralList();
|
||||
|
||||
final foreignKeyBranches = foreignKeys.indexed.map((e) {
|
||||
final (index, foreignKey) = e;
|
||||
final childColumn = foreignKey.columns.single.escapeIdentifier();
|
||||
final parentTable = foreignKey.referenceTable.escapeIdentifier();
|
||||
final parentColumn = foreignKey.referenceColumns.single.escapeIdentifier();
|
||||
final (parentTableId, _) = schema[foreignKey.referenceTable]!;
|
||||
|
||||
return '''
|
||||
SELECT $index AS fk_idx, c."$childColumn" AS invalid_value
|
||||
FROM "$escapedChildTable" c
|
||||
LEFT JOIN "$parentTable" p
|
||||
ON p."$parentColumn" = c."$childColumn"
|
||||
LEFT JOIN "crdt_data_rows" r
|
||||
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
|
||||
p."scopeId" <> $scopeId OR
|
||||
r."visibility" > $crdtRowLastVisibleVisibilityIndex
|
||||
)
|
||||
''';
|
||||
});
|
||||
|
||||
final result = await database.unsafeQuery(
|
||||
'${foreignKeyBranches.join('UNION ALL\n')}LIMIT 1',
|
||||
transaction: transaction,
|
||||
);
|
||||
if (result.isEmpty) return null;
|
||||
|
||||
return (
|
||||
foreignKeyIndex: result.first[0] as int,
|
||||
value: result.first[1] as Object,
|
||||
);
|
||||
}
|
||||
|
||||
/// The unique-index conflict release metadata for [tableDefinition].
|
||||
List<UniqueIndexConflictRelease> uniqueIndexesForTable(
|
||||
TableDefinition tableDefinition,
|
||||
) => databaseContext.uniqueIndexesForTable(tableDefinition);
|
||||
}
|
||||
|
||||
int _nextClFlag(CrdtDataDeleted? current, bool isDeleted) {
|
||||
var next = (current?.clFlag ?? 1) + 1;
|
||||
if (next.isEven != isDeleted) next++;
|
||||
return next;
|
||||
}
|
||||
|
|
@ -0,0 +1,31 @@
|
|||
import 'package:meta/meta.dart';
|
||||
import 'package:serverpod_serialization/serverpod_serialization.dart';
|
||||
|
||||
import '../../crdt/sync.dart';
|
||||
import '../../generated/protocol.dart';
|
||||
import '../../hlc/hlc.dart';
|
||||
|
||||
/// Identifies a domain row during merges: `(tableName, rowId)`.
|
||||
@internal
|
||||
typedef MergeRowKey = (String, UuidValue);
|
||||
|
||||
/// Identifies a domain field during merges: `(tableName, rowId, columnName)`.
|
||||
@internal
|
||||
typedef MergeFieldKey = (String, UuidValue, String);
|
||||
|
||||
/// Preloaded CRDT metadata shared while applying a merge set.
|
||||
@internal
|
||||
typedef MergeContext = ({
|
||||
Map<MergeRowKey, CrdtDataRow> rows,
|
||||
Map<MergeFieldKey, CrdtDataField> fields,
|
||||
Map<MergeFieldKey, Hlc> incomingFieldHlcs,
|
||||
Map<MergeRowKey, CrdtDataDeleted> tombstones,
|
||||
Map<MergeRowKey, DomainRowOwner> domainOwners,
|
||||
});
|
||||
|
||||
/// Remote nodes and scope-node rows loaded for a merge.
|
||||
@internal
|
||||
typedef MergeNodes = ({
|
||||
Map<UuidValue, CrdtNode> nodesByUuid,
|
||||
Map<UuidValue, CrdtScopeNode> scopeNodesByUuid,
|
||||
});
|
||||
|
|
@ -1,22 +1,35 @@
|
|||
part of '../recorder.dart';
|
||||
import 'package:meta/meta.dart';
|
||||
import 'package:serverpod_database/serverpod_database.dart';
|
||||
import 'package:serverpod_serialization/serverpod_serialization.dart';
|
||||
|
||||
import '../../crdt/extensions.dart';
|
||||
import '../../crdt/merge.dart';
|
||||
import '../../generated/protocol.dart';
|
||||
import '../../hlc/hlc.dart';
|
||||
import '../unique_index_utils.dart';
|
||||
import 'database_helpers.dart';
|
||||
import 'recorder_context.dart';
|
||||
import 'types.dart';
|
||||
|
||||
typedef _UniqueConflict = ({
|
||||
CrdtDataRow row,
|
||||
_UniqueIndexConflictRelease uniqueIndex,
|
||||
UniqueIndexConflictRelease uniqueIndex,
|
||||
});
|
||||
|
||||
/// Resolves unique constraint conflicts that appear while merging CRDT changes.
|
||||
@internal
|
||||
/// Resolves unique constraint conflicts that appear while merging CRDT
|
||||
/// changes or soft-deleting rows.
|
||||
class CrdtUniqueConflictResolver {
|
||||
/// Creates a resolver for the merge recorder.
|
||||
CrdtUniqueConflictResolver(this._recorder);
|
||||
/// Creates a resolver over the given recorder context.
|
||||
CrdtUniqueConflictResolver(this._context);
|
||||
|
||||
final CrdtMutationRecorder _recorder;
|
||||
final CrdtRecorderContext _context;
|
||||
|
||||
/// Resolves unique conflicts for an incoming insert.
|
||||
Future<({Map<String, Object?> data, bool changed})> _resolveForIncomingInsert(
|
||||
Future<({Map<String, Object?> data, bool changed})> resolveForIncomingInsert(
|
||||
CrdtMergeInsert insert,
|
||||
Map<String, Object?> data,
|
||||
_MergeContext context,
|
||||
MergeContext context,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final resolvedData = Map<String, Object?>.from(data);
|
||||
|
|
@ -67,8 +80,8 @@ class CrdtUniqueConflictResolver {
|
|||
|
||||
Hlc _incomingInsertUniqueClaim({
|
||||
required CrdtMergeInsert insert,
|
||||
required _UniqueIndexConflictRelease uniqueIndex,
|
||||
required _MergeContext context,
|
||||
required UniqueIndexConflictRelease uniqueIndex,
|
||||
required MergeContext context,
|
||||
}) {
|
||||
var uniqueHlc = insert.hlc;
|
||||
for (final columnName in uniqueIndex.indexedColumns) {
|
||||
|
|
@ -84,13 +97,13 @@ class CrdtUniqueConflictResolver {
|
|||
}
|
||||
|
||||
/// Resolves unique conflicts for a single incoming field update.
|
||||
Future<Map<String, Object?>> _resolveForIncomingUpdate(
|
||||
Future<Map<String, Object?>> resolveForIncomingUpdate(
|
||||
CrdtMergeUpdate update,
|
||||
CrdtDataRow row,
|
||||
_MergeContext context,
|
||||
MergeContext context,
|
||||
Transaction transaction,
|
||||
) {
|
||||
return _resolveForIncomingUpdates(
|
||||
return resolveForIncomingUpdates(
|
||||
tableName: update.tableName,
|
||||
row: row,
|
||||
updates: {update.columnName: update.value},
|
||||
|
|
@ -100,11 +113,11 @@ class CrdtUniqueConflictResolver {
|
|||
}
|
||||
|
||||
/// Resolves unique conflicts for one or more incoming field updates.
|
||||
Future<Map<String, Object?>> _resolveForIncomingUpdates({
|
||||
Future<Map<String, Object?>> resolveForIncomingUpdates({
|
||||
required String tableName,
|
||||
required CrdtDataRow row,
|
||||
required Map<String, Object?> updates,
|
||||
required _MergeContext context,
|
||||
required MergeContext context,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
if (updates.keys.every(
|
||||
|
|
@ -113,20 +126,20 @@ class CrdtUniqueConflictResolver {
|
|||
return updates;
|
||||
}
|
||||
|
||||
final tableDefinition = _recorder._tableDefinitionsByName[tableName];
|
||||
final tableDefinition = _context.tableDefinitionsByName[tableName];
|
||||
if (tableDefinition == null) {
|
||||
return updates;
|
||||
}
|
||||
|
||||
final uniqueColumnNamesToRead = {
|
||||
for (final uniqueIndex in _recorder._uniqueIndexesForTable(tableDefinition))
|
||||
for (final uniqueIndex in _context.uniqueIndexesForTable(tableDefinition))
|
||||
...uniqueIndex.indexedColumns,
|
||||
};
|
||||
if (uniqueColumnNamesToRead.isEmpty) {
|
||||
return updates;
|
||||
}
|
||||
|
||||
final currentValues = await _recorder._readDomainColumnValues(
|
||||
final currentValues = await _context.readDomainColumnValues(
|
||||
tableName,
|
||||
{row.uuidRowId},
|
||||
uniqueColumnNamesToRead.toList(),
|
||||
|
|
@ -191,9 +204,9 @@ class CrdtUniqueConflictResolver {
|
|||
String tableName,
|
||||
UuidValue rowId,
|
||||
Map<String, Object?> data,
|
||||
_UniqueIndexConflictRelease uniqueIndex,
|
||||
UniqueIndexConflictRelease uniqueIndex,
|
||||
) {
|
||||
final tableDefinition = _recorder._tableDefinitionsByName[tableName];
|
||||
final tableDefinition = _context.tableDefinitionsByName[tableName];
|
||||
if (tableDefinition == null) return data;
|
||||
|
||||
final released = Map<String, Object?>.from(data);
|
||||
|
|
@ -204,7 +217,7 @@ class CrdtUniqueConflictResolver {
|
|||
if (values.values.any((value) => value == null)) return released;
|
||||
|
||||
for (final column in uniqueIndex.releaseColumns) {
|
||||
released[column.columnName] = _recorder._conflictFreeValue(
|
||||
released[column.columnName] = _context.conflictFreeValue(
|
||||
column,
|
||||
values[column.columnName],
|
||||
tableName,
|
||||
|
|
@ -218,10 +231,10 @@ class CrdtUniqueConflictResolver {
|
|||
Future<void> _releaseUniqueConflictForRow({
|
||||
required String tableName,
|
||||
required UuidValue rowId,
|
||||
required _UniqueIndexConflictRelease uniqueIndex,
|
||||
required UniqueIndexConflictRelease uniqueIndex,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
final valuesByRowId = await _recorder._readDomainColumnValues(
|
||||
final valuesByRowId = await _context.readDomainColumnValues(
|
||||
tableName,
|
||||
{rowId},
|
||||
uniqueIndex.indexedColumns,
|
||||
|
|
@ -236,7 +249,7 @@ class CrdtUniqueConflictResolver {
|
|||
values,
|
||||
uniqueIndex,
|
||||
);
|
||||
await _recorder._updateDomainRows(tableName, {rowId}, releasedValues, transaction);
|
||||
await _context.updateDomainRows(tableName, {rowId}, releasedValues, transaction);
|
||||
}
|
||||
|
||||
Future<List<_UniqueConflict>> _findVisibleUniqueConflicts({
|
||||
|
|
@ -245,22 +258,20 @@ class CrdtUniqueConflictResolver {
|
|||
required Map<String, Object?> values,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
final tableDefinition = _recorder._tableDefinitionsByName[tableName];
|
||||
final tableDefinition = _context.tableDefinitionsByName[tableName];
|
||||
if (tableDefinition == null) return const [];
|
||||
|
||||
final uniqueIndexes = _recorder._uniqueIndexesForTable(tableDefinition);
|
||||
final uniqueIndexes = _context.uniqueIndexesForTable(tableDefinition);
|
||||
if (uniqueIndexes.isEmpty) return const [];
|
||||
|
||||
final uniqueIndexesByConflictId = <UuidValue, Set<_UniqueIndexConflictRelease>>{};
|
||||
final uniqueIndexesByConflictId = <UuidValue, Set<UniqueIndexConflictRelease>>{};
|
||||
for (final uniqueIndex in uniqueIndexes) {
|
||||
final columnValues = {
|
||||
for (final columnName in uniqueIndex.indexedColumns)
|
||||
if (values.containsKey(columnName)) columnName: values[columnName],
|
||||
};
|
||||
if (uniqueIndex.scoped) {
|
||||
columnValues['scopeId'] = _recorder
|
||||
._getHlcManager(transaction)
|
||||
.normalizedScopeId;
|
||||
columnValues['scopeId'] = _context.hlcManagerFor(transaction).normalizedScopeId;
|
||||
}
|
||||
final expectedValueCount =
|
||||
uniqueIndex.indexedColumns.length + (uniqueIndex.scoped ? 1 : 0);
|
||||
|
|
@ -279,7 +290,7 @@ class CrdtUniqueConflictResolver {
|
|||
}
|
||||
|
||||
if (uniqueIndexesByConflictId.isEmpty) return const [];
|
||||
final conflictRows = await _recorder._findCrdtRows(
|
||||
final conflictRows = await _context.findCrdtRows(
|
||||
tableName,
|
||||
uniqueIndexesByConflictId.keys.toSet(),
|
||||
transaction,
|
||||
|
|
@ -307,8 +318,8 @@ class CrdtUniqueConflictResolver {
|
|||
Future<Hlc> _uniqueIndexHlc({
|
||||
required String tableName,
|
||||
required CrdtDataRow row,
|
||||
required _UniqueIndexConflictRelease uniqueIndex,
|
||||
required Map<_MergeFieldKey, CrdtDataField> fields,
|
||||
required UniqueIndexConflictRelease uniqueIndex,
|
||||
required Map<MergeFieldKey, CrdtDataField> fields,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
final uniqueColumnNames = {
|
||||
|
|
@ -319,7 +330,7 @@ class CrdtUniqueConflictResolver {
|
|||
for (final columnName in uniqueColumnNames) {
|
||||
final field = fields[(tableName, row.uuidRowId, columnName)];
|
||||
if (field == null) {
|
||||
final columnId = _schemaColumn(_recorder._schema, tableName, columnName)?.id;
|
||||
final columnId = _context.schemaColumn(tableName, columnName)?.id;
|
||||
if (columnId != null) missingColumnIds.add(columnId);
|
||||
continue;
|
||||
}
|
||||
|
|
@ -329,7 +340,7 @@ class CrdtUniqueConflictResolver {
|
|||
|
||||
if (missingColumnIds.isNotEmpty && row.id != null) {
|
||||
final loadedFields = await CrdtDataField.db.find(
|
||||
_recorder._session,
|
||||
_context.databaseSession,
|
||||
where: (t) => t.rowId.equals(row.id) & t.columnId.inSet(missingColumnIds),
|
||||
include: CrdtDataField.include(
|
||||
column: CrdtSchemaColumn.include(),
|
||||
|
|
@ -356,28 +367,71 @@ class CrdtUniqueConflictResolver {
|
|||
required Map<String, Object?> values,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
return _recorder._findVisibleDomainRowIdsWhere(
|
||||
return _context.findVisibleDomainRowIdsWhere(
|
||||
tableName: tableName,
|
||||
predicates: [
|
||||
for (final MapEntry(key: columnName, value: value) in values.entries)
|
||||
_domainColumnPredicate(columnName, value),
|
||||
_domainColumnNotPredicate('id', rowId),
|
||||
domainColumnPredicate(columnName, value),
|
||||
domainColumnNotPredicate('id', rowId),
|
||||
],
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
bool _isUniqueIndexedColumn(String tableName, String columnName) {
|
||||
final tableDefinition = _recorder._tableDefinitionsByName[tableName];
|
||||
final tableDefinition = _context.tableDefinitionsByName[tableName];
|
||||
if (tableDefinition == null) return false;
|
||||
return _recorder
|
||||
._uniqueIndexesForTable(tableDefinition)
|
||||
return _context
|
||||
.uniqueIndexesForTable(tableDefinition)
|
||||
.any((i) => i.indexedColumns.contains(columnName));
|
||||
}
|
||||
}
|
||||
|
||||
extension on _UniqueIndexConflictRelease {
|
||||
Set<String> get releaseColumnNames => {
|
||||
for (final column in releaseColumns) column.columnName,
|
||||
};
|
||||
Future<void> releaseOnDelete(
|
||||
String tableName,
|
||||
Set<UuidValue> rowIds,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final tableDefinition = _context.tableDefinitionsByName[tableName];
|
||||
if (tableDefinition == null) return;
|
||||
|
||||
final uniqueIndexes = _context.uniqueIndexesForTable(tableDefinition);
|
||||
for (final uniqueIndex in uniqueIndexes) {
|
||||
final releaseColumnNames = uniqueIndex.releaseColumnNames.toList();
|
||||
|
||||
final valuesByRowId = await _context.readDomainColumnValues(
|
||||
tableName,
|
||||
rowIds,
|
||||
uniqueIndex.indexedColumns,
|
||||
transaction,
|
||||
);
|
||||
final updatedRowIds = <UuidValue>{};
|
||||
|
||||
for (final MapEntry(key: rowId, value: values) in valuesByRowId.entries) {
|
||||
if (values.values.any((value) => value == null)) continue;
|
||||
|
||||
final updates = <String, Object?>{};
|
||||
for (final column in uniqueIndex.releaseColumns) {
|
||||
updates[column.columnName] = _context.conflictFreeValue(
|
||||
column,
|
||||
values[column.columnName],
|
||||
tableDefinition.name,
|
||||
rowId,
|
||||
'deleted',
|
||||
);
|
||||
}
|
||||
|
||||
await _context.updateDomainRows(tableName, {rowId}, updates, transaction);
|
||||
updatedRowIds.add(rowId);
|
||||
}
|
||||
|
||||
if (updatedRowIds.isNotEmpty) {
|
||||
await _context.recordFieldsUpdatedByTable(
|
||||
tableName,
|
||||
updatedRowIds,
|
||||
releaseColumnNames,
|
||||
transaction,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -2,6 +2,7 @@ import 'package:serverpod_database/serverpod_database.dart';
|
|||
import 'package:uuid/uuid.dart';
|
||||
|
||||
import '../generated/protocol.dart';
|
||||
import 'merge_utils/database_helpers.dart';
|
||||
import 'unique_index_utils.dart';
|
||||
|
||||
/// Manages the CRDT schema for a database.
|
||||
|
|
@ -23,10 +24,9 @@ class CrdtSchemaRegistry {
|
|||
);
|
||||
}
|
||||
|
||||
final tablesWithoutScopeId = syncTables.where((table) {
|
||||
final scopeColumn = _scopeIdColumn(table);
|
||||
return scopeColumn == null || scopeColumn is! Column<int>;
|
||||
}).toList();
|
||||
final tablesWithoutScopeId = syncTables
|
||||
.where((table) => table.crdtScopeIdColumn == null)
|
||||
.toList();
|
||||
if (tablesWithoutScopeId.isNotEmpty) {
|
||||
throw StateError(
|
||||
'CRDT can only synchronize tables with a nullable int scopeId column, '
|
||||
|
|
@ -212,13 +212,6 @@ class CrdtSchemaRegistry {
|
|||
}
|
||||
}
|
||||
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ extension _CrdtDatabaseScope on CrdtDatabase {
|
|||
for (final (index, row) in rows.indexed) {
|
||||
final rowScopeId = _readScopeId(row);
|
||||
if (rowScopeId == null) {
|
||||
databaseRows.add(_copyWithScopeId(row, effectiveScopeId));
|
||||
databaseRows.add(row.copyWithScopeId(effectiveScopeId));
|
||||
stampedIndexes.add(index);
|
||||
continue;
|
||||
}
|
||||
|
|
@ -89,10 +89,6 @@ extension _CrdtDatabaseScope on CrdtDatabase {
|
|||
|
||||
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,
|
||||
({
|
||||
|
|
@ -210,8 +206,7 @@ extension ScopeUuidFilter on Table {
|
|||
);
|
||||
}
|
||||
|
||||
Column get _scopeIdColumn => columns.singleWhere(
|
||||
(column) => column.columnName == 'scopeId' && column is ColumnInt,
|
||||
orElse: () => throw StateError('Table "$tableName" has no scopeId column.'),
|
||||
);
|
||||
Column get _scopeIdColumn =>
|
||||
crdtScopeIdColumn ??
|
||||
(throw StateError('Table "$tableName" has no scopeId column.'));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import 'package:serverpod_database/serverpod_database.dart';
|
|||
|
||||
import '../crdt/extensions.dart';
|
||||
import '../generated/protocol.dart';
|
||||
import 'merge_utils/database_helpers.dart';
|
||||
|
||||
/// Resolves the local [CrdtSchemaTable] id for a domain table name. Returns
|
||||
/// null when the table is not registered for CRDT synchronization.
|
||||
|
|
@ -204,7 +205,7 @@ extension on Table {
|
|||
// Synced tables are validated to carry the column at initialize(); a
|
||||
// missing column here must fail closed, never skip the predicate.
|
||||
final scopeColumn =
|
||||
scopeIdColumn ??
|
||||
crdtScopeIdColumn ??
|
||||
(throw StateError(
|
||||
'Synced table "$tableName" has no int scopeId column; '
|
||||
'CrdtSchemaRegistry validation should have rejected it.',
|
||||
|
|
@ -218,7 +219,7 @@ extension on Table {
|
|||
|
||||
final scopeFilter = effectiveScopeIds == null
|
||||
? '${crdtRow.scopeId} = $scopeColumn'
|
||||
: '${crdtRow.scopeId} IN (${_sqlIntList(effectiveScopeIds)})';
|
||||
: '${crdtRow.scopeId} IN (${effectiveScopeIds.sqlLiteralList()})';
|
||||
final notExistsExpr = Expression(
|
||||
'NOT EXISTS '
|
||||
'(SELECT 1 FROM "${crdtRow.tableName}" '
|
||||
|
|
@ -233,17 +234,4 @@ extension on Table {
|
|||
? (scopeColumn.inSet(effectiveScopeIds.toSet()) & notExistsExpr)
|
||||
: notExistsExpr);
|
||||
}
|
||||
|
||||
ColumnInt? get scopeIdColumn {
|
||||
for (final column in columns) {
|
||||
if (column.columnName == 'scopeId' && column is ColumnInt) {
|
||||
return column;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
String _sqlIntList(Iterable<int> values) {
|
||||
return values.map(ValueEncoder.instance.convert).join(', ');
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,69 @@
|
|||
import 'package:meta/meta.dart';
|
||||
import 'package:serverpod_database/serverpod_database.dart';
|
||||
|
||||
@internal
|
||||
typedef UniqueIndexConflictRelease = ({
|
||||
List<String> indexedColumns,
|
||||
List<UniqueColumnConflictRelease> releaseColumns,
|
||||
bool scoped,
|
||||
});
|
||||
|
||||
@internal
|
||||
typedef UniqueColumnConflictRelease = ({
|
||||
String columnName,
|
||||
CrdtUniqueConflictReleaseKind kind,
|
||||
});
|
||||
|
||||
@internal
|
||||
List<UniqueIndexConflictRelease> syncableUniqueIndexesForTable(
|
||||
TableDefinition table,
|
||||
Set<String> syncTableNames,
|
||||
) {
|
||||
final columnsByName = {for (final column in table.columns) column.name: column};
|
||||
return [
|
||||
for (final index in crdtSyncableUniqueIndexesForTable(table, syncTableNames))
|
||||
uniqueIndexConflictReleaseForIndex(table, columnsByName, index),
|
||||
];
|
||||
}
|
||||
|
||||
@internal
|
||||
UniqueIndexConflictRelease uniqueIndexConflictReleaseForIndex(
|
||||
TableDefinition table,
|
||||
Map<String, ColumnDefinition> columnsByName,
|
||||
IndexDefinition index,
|
||||
) {
|
||||
final columnNames = index.elements.map((element) => element.definition).toList();
|
||||
final indexedColumns = [
|
||||
for (final columnName in columnNames)
|
||||
if (columnName != 'scopeId') columnName,
|
||||
];
|
||||
final releaseColumns = <UniqueColumnConflictRelease>[];
|
||||
for (final columnName in indexedColumns) {
|
||||
final column = columnsByName[columnName];
|
||||
if (column == null) {
|
||||
throw StateError('No column definition found for ${table.name}.$columnName.');
|
||||
}
|
||||
final kind = crdtUniqueConflictReleaseKindForColumn(table, column);
|
||||
if (kind != null) {
|
||||
releaseColumns.add((columnName: columnName, kind: kind));
|
||||
}
|
||||
}
|
||||
|
||||
return (
|
||||
scoped: columnNames.contains('scopeId'),
|
||||
indexedColumns: indexedColumns,
|
||||
releaseColumns: releaseColumns,
|
||||
);
|
||||
}
|
||||
|
||||
@internal
|
||||
extension UniqueIndexConflictReleaseExtension on UniqueIndexConflictRelease {
|
||||
/// Column names released when tombstoning rows covered by this index.
|
||||
Set<String> get releaseColumnNames => {
|
||||
for (final column in releaseColumns) column.columnName,
|
||||
};
|
||||
}
|
||||
|
||||
@internal
|
||||
bool isCrdtAllowedForeignKeyOnlyUniqueIndex(
|
||||
TableDefinition tableDefinition,
|
||||
|
|
|
|||
Loading…
Reference in a new issue