feat: Implements foreign key handling (#68)

This commit is contained in:
Marcelo Mendonça Soares 2026-06-10 17:31:05 -03:00 • committed by GitHub
commit 4580326956
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 6630 additions and 189 deletions

View file

@ -2,6 +2,21 @@ import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart
import '../protocol/protocol.dart';
/// Last serialized index that still represents a visible CRDT row.
///
/// The [CrdtDataRowVisibility] is ordered by visibility, so the index is used
/// to compare the visibility states. The last visible state in the enum is
/// [CrdtDataRowVisibility.foreignKeyRestrictRestore].
final crdtRowLastVisibleVisibilityIndex =
CrdtDataRowVisibility.foreignKeyRestrictRestore.index;
/// Extensions for the [CrdtDataRowVisibility] enum.
extension CrdtDataRowVisibilityExtension on CrdtDataRowVisibility {
/// Whether this visibility state hides the row from domain queries.
///
bool get isHidden => index > crdtRowLastVisibleVisibilityIndex;
}
/// Extensions for the [CrdtDataField] class.
extension CrdtDataFieldExtension on CrdtDataField {
/// The HLC represented by this field.
@ -14,6 +29,9 @@ extension CrdtDataFieldExtension on CrdtDataField {
/// Extensions for the [CrdtDataRow] class.
extension CrdtDataRowExtension on CrdtDataRow {
/// Whether this row is currently hidden from domain queries.
bool get isHidden => visibility.isHidden;
/// The HLC represented by this row.
Hlc get hlc => node == null
? (throw StateError(
@ -34,3 +52,20 @@ extension CrdtDataDeletedExtension on CrdtDataDeleted {
))
: toHlcForNode(node!.uuidNodeId);
}
/// Extensions for the [CrdtDataForeignKey] class.
extension CrdtDataForeignKeyExtension on CrdtDataForeignKey {
/// Whether an active projection override exists for this FK field.
///
/// Derived from [CrdtDataForeignKey.overrideReason]: an override is active if
/// and only if [overrideReason] is non-null. This is the authoritative test
/// for whether the domain row's FK value differs from the attempted value.
bool get hasOverride => overrideReason != null;
}
/// Extensions for the [CrdtDataDeletedReason] enum.
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');
}

View file

@ -9,6 +9,7 @@ import '../managers/user.dart';
import '../protocol/protocol.dart';
import '../utils/case_when.dart' show Case;
import 'exceptions.dart';
import 'extensions.dart';
import 'merge.dart';
/// Callback function for when a merge is successful.
@ -128,6 +129,11 @@ class CrdtSync {
/// Changes are emitted in insert, update, then delete order. Domain row and
/// column payloads are resolved incrementally as each change is yielded.
///
/// Foreign-key columns with an active projection override are sent with their
/// durable [CrdtDataForeignKey.attemptedValue], not the visible value stored
/// in the domain table. Peers need the attempted fact to converge; local FK
/// projection materializes only the safe visible value into domain tables.
///
/// All changes for nodes that are not present in the [nodeCheckpoints] list
/// are collected and emitted. Passing an empty list will collect all changes.
Stream<CrdtMergeChange> collectPendingChanges(
@ -171,6 +177,9 @@ class CrdtSync {
/// Merges a remote [mergeSet] and records the sync checkpoint for [otherNodeId].
///
/// Inbound merge applies each remote change, then materializes foreign-key
/// projection into domain tables via [CrdtDatabase.mergeChanges].
///
/// Throws if the merge fails. The sync stream should be closed so the next
/// attempt resumes from the last persisted checkpoint.
///
@ -368,6 +377,11 @@ class CrdtSync {
),
);
final foreignKeyAttemptFieldsByRowId = await _loadProjectedForeignKeyAttemptFields(
session,
rows,
);
for (final row in rows) {
final tableName = row.tbl!.name;
if (!_syncTablesByName.containsKey(tableName)) continue;
@ -382,6 +396,7 @@ class CrdtSync {
row.uuidRowId,
table,
dartName,
foreignKeyAttemptFieldsByRowId[row.id!],
);
if (domainRow == null) continue;
@ -408,12 +423,18 @@ class CrdtSync {
row: CrdtDataRow.include(tbl: CrdtSchemaTable.include()),
column: CrdtSchemaColumn.include(),
node: CrdtNode.include(),
foreignKey: CrdtDataForeignKey.include(),
),
);
for (final field in fields) {
final tableName = field.row!.tbl!.name;
if (!_syncTablesByName.containsKey(tableName)) continue;
if (field.hlcDatetime == field.row!.hlcDatetime &&
field.hlcCounter == field.row!.hlcCounter &&
field.nodeId == field.row!.nodeId) {
continue;
}
final columnName = field.column!.name;
final decodedValue = await _fetchColumnValue(
@ -421,6 +442,7 @@ class CrdtSync {
tableName,
field.row!.uuidRowId,
columnName,
field.foreignKey,
);
yield CrdtMergeUpdate(
@ -450,6 +472,8 @@ class CrdtSync {
);
for (final tombstone in tombstones) {
if (!tombstone.reason.isSynced) continue;
final tableName = tombstone.row!.tbl!.name;
if (!_syncTablesByName.containsKey(tableName)) continue;
@ -504,41 +528,97 @@ class CrdtSync {
nodeCheckpoints,
);
/// Loads a domain row for outbound insert sync.
///
/// Reads the materialized row from the domain table, then swaps any FK columns
/// 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(
DatabaseSession session,
String tableName,
UuidValue rowId,
Table table,
String dartName,
List<CrdtDataField>? foreignKeyAttemptFields,
) async {
final cols = table.columns.map((column) => '"${column.columnName}"').join(', ');
final cols = table.columns
.map((column) => '"${_escapeIdentifier(column.columnName)}"')
.join(', ');
final encodedRowId = ValueEncoder.instance.convert(rowId);
final result = await session.db.unsafeQuery(
'SELECT $cols FROM "$tableName" WHERE "id" = $encodedRowId',
'SELECT $cols FROM "${_escapeIdentifier(tableName)}" '
'WHERE "id" = $encodedRowId',
);
if (result.isEmpty) return null;
final columnMap = result.first.toColumnMap();
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);
return session.db.serializationManager.deserializeByClassName({
'className': dartName,
'data': columnMap,
});
}
/// Resolves a column value for outbound update sync.
///
/// When [projection] has an active override, returns the attempted value
/// 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(
DatabaseSession session,
String tableName,
UuidValue rowId,
String columnName,
CrdtDataForeignKey? projection,
) async {
if (projection != null && projection.hasOverride) {
return _decodeColumnValue(tableName, columnName, projection.attemptedValue);
}
final encodedValue = ValueEncoder.instance.convert(rowId);
final result = await session.db.unsafeQuery(
'SELECT "$columnName" FROM "$tableName" WHERE "id" = $encodedValue',
'SELECT "${_escapeIdentifier(columnName)}" '
'FROM "${_escapeIdentifier(tableName)}" '
'WHERE "id" = $encodedValue',
);
if (result.isEmpty) return null;
return _decodeColumnValue(tableName, columnName, result.first[0]);
}
/// Loads FK projection metadata for outbound insert sync.
///
/// Returns only fields whose [CrdtDataForeignKey.overrideReason] is non-null,
/// keyed by CRDT row id. These are the columns whose domain-table value
/// differs from the durable attempted FK fact.
Future<Map<int, List<CrdtDataField>>> _loadProjectedForeignKeyAttemptFields(
DatabaseSession session,
List<CrdtDataRow> rows,
) async {
final rowIds = {for (final row in rows) ?row.id};
if (rowIds.isEmpty) return {};
final fields = await CrdtDataField.db.find(
session,
where: (t) => t.rowId.inSet(rowIds) & t.foreignKey.overrideReason.notEquals(null),
include: CrdtDataField.include(
column: CrdtSchemaColumn.include(),
foreignKey: CrdtDataForeignKey.include(),
),
);
final fieldsByRowId = <int, List<CrdtDataField>>{};
for (final field in fields) {
fieldsByRowId.putIfAbsent(field.rowId, () => []).add(field);
}
return fieldsByRowId;
}
dynamic _decodeColumnValue(String tableName, String columnName, Object? value) {
if (value == null) return null;
@ -563,6 +643,8 @@ class CrdtSync {
return withoutNullable.split(':').last;
}
String _escapeIdentifier(String identifier) => identifier.replaceAll('"', '""');
static String _computeCanonicalSyncTablesSignature(
List<Table> syncTables, {
required List<TableDefinition> tableDefinitions,
@ -650,3 +732,21 @@ extension on CrdtNodeTable {
return caseExpression.orElse(Constant.bool(true));
}
}
extension on Map<String, dynamic> {
/// Replaces materialized FK column values with attempted values for sync.
///
/// After local FK projection (for example `SET NULL` or `SET DEFAULT`), the
/// domain table stores the safe visible value while
/// [CrdtDataForeignKey.attemptedValue] preserves what was actually tried.
/// Outbound sync must send the attempted value so peers can apply their own
/// projection from the same fact.
void applyProjectedForeignKeyAttempts(List<CrdtDataField>? foreignKeyAttemptFields) {
if (foreignKeyAttemptFields == null) return;
for (final field in foreignKeyAttemptFields) {
final projection = field.foreignKey;
if (projection == null || !projection.hasOverride) continue;
this[field.column!.name] = projection.attemptedValue;
}
}
}

View file

@ -296,10 +296,13 @@ class CrdtDatabase implements Database {
_delegate,
transaction,
(tx) async {
return [
final updatedRows = [
for (final row in rows)
await updateRow<T>(row, columns: columns, transaction: tx),
await _updateRowWithoutRecording(row, columns: columns, transaction: tx),
];
await _recorder.afterUpdate(updatedRows, columns, tx);
return updatedRows;
},
);
}
@ -309,6 +312,27 @@ class CrdtDatabase implements Database {
T row, {
List<Column>? columns,
Transaction? transaction,
}) async {
return DatabaseUtil.runInTransactionOrSavepoint(
_delegate,
transaction,
(tx) async {
final updatedRow = await _updateRowWithoutRecording(
row,
columns: columns,
transaction: tx,
);
await _recorder.afterUpdate([updatedRow], columns, tx);
return updatedRow;
},
);
}
Future<T> _updateRowWithoutRecording<T extends TableRow>(
T row, {
required Transaction transaction,
List<Column>? columns,
}) async {
final values = row.toJsonForDatabase() as Map<String, dynamic>;
final columnValues = (columns ?? row.table.managedColumns)
@ -316,9 +340,10 @@ class CrdtDatabase implements Database {
.map((c) => ColumnValue(c, values[c.columnName]))
.toList();
final updatedRows = await updateWhere<T>(
final where = row.table.id.equals(row.id);
final updatedRows = await _delegate.updateWhere<T>(
columnValues: columnValues,
where: row.table.id.equals(row.id),
where: mergeWhereWithTombstone<T>(serializationManager, where, null)!,
transaction: transaction,
);

View file

@ -36,6 +36,9 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
if (mergeSet.isEmpty) return;
final operations = mergeSet.causallyOrderedChanges;
final shouldProjectForeignKeys = _mergeOperationsMayAffectForeignKeys(
operations,
);
final currentUser = _getEffectiveUser(transaction);
final remoteNodes = await _findOrCreateNodesForMerge(
currentUser.id!,
@ -74,6 +77,9 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
}
}
if (shouldProjectForeignKeys) {
await _projectForeignKeys(transaction);
}
await _updateHlcFromIncomingOperations(
operations,
remoteNodes,
@ -221,16 +227,13 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
);
}
final (_, columnsByName) = _schema[tableName]!;
final columnIds = <int>{
for (final columnName in columnNames)
if (columnsByName[columnName] != null) columnsByName[columnName]!.id!,
};
if (columnIds.isEmpty) continue;
final columnIdsByName = _schemaColumnIds(_schema, tableName, columnNames);
if (columnIdsByName.isEmpty) continue;
final loadedFields = await CrdtDataField.db.find(
_session,
where: (t) => t.rowId.inSet(rowPks) & t.columnId.inSet(columnIds),
where: (t) =>
t.rowId.inSet(rowPks) & t.columnId.inSet(columnIdsByName.values.toSet()),
include: CrdtDataField.include(
column: CrdtSchemaColumn.include(),
node: CrdtNode.include(),
@ -277,13 +280,26 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
transaction,
);
final safeInsert = await _safeIncomingForeignKeyData(
insert.tableName,
insert.uuidRowId,
resolvedInsert.data,
transaction,
);
context.rows[rowKey] = await _applyMergeInsertForMissingRow(
insert,
remoteNode,
incomingHlc,
resolvedInsert.data,
safeInsert.data,
transaction,
);
await _recordForeignKeyInsertAttempts(
insert.tableName,
{insert.uuidRowId},
transaction,
safeInsert.attempts,
);
} else {
await _applyMergeInsertForExistingRow(
insert,
@ -326,12 +342,25 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
context,
transaction,
);
final safeUpdate = await _safeIncomingForeignKeyData(
update.tableName,
update.uuidRowId,
resolvedUpdates,
transaction,
);
await _updateDomainRows(
update.tableName,
{update.uuidRowId},
resolvedUpdates,
safeUpdate.data,
transaction,
);
await _recordForeignKeyAttemptsForRows(
update.tableName,
{update.uuidRowId},
resolvedUpdates.keys.toSet(),
transaction,
attemptedValues: safeUpdate.attempts,
);
}
Future<void> _applyMergeDelete(
@ -343,6 +372,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
final rowKey = (delete.tableName, delete.uuidRowId);
final row = context.rows[rowKey];
if (row == null) return;
if (!delete.reason.isSynced) return;
final currentTombstone = context.tombstones[rowKey];
final currentClFlag = currentTombstone?.clFlag ?? 1;
@ -359,6 +389,14 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
currentTombstone,
transaction,
);
if (delete.reason.isSynced) {
await _applyRowVisibilityFromUserTombstone(
row,
context.tombstones[rowKey]!,
transaction,
);
}
}
Future<CrdtDataRow> _upsertMergeRow(
@ -575,12 +613,25 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
}
if (updatesToApply.isNotEmpty) {
final safeUpdate = await _safeIncomingForeignKeyData(
insert.tableName,
insert.uuidRowId,
updatesToApply,
transaction,
);
await _updateDomainRows(
insert.tableName,
{insert.uuidRowId},
updatesToApply,
safeUpdate.data,
transaction,
);
await _recordForeignKeyAttemptsForRows(
insert.tableName,
{insert.uuidRowId},
updatesToApply.keys.toSet(),
transaction,
attemptedValues: safeUpdate.attempts,
);
}
final rowKey = (insert.tableName, insert.uuidRowId);
@ -596,6 +647,11 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
currentTombstone,
transaction,
);
await _applyRowVisibilityFromUserTombstone(
currentRow,
context.tombstones[rowKey]!,
transaction,
);
}
}
@ -610,8 +666,8 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
required Transaction transaction,
CrdtSchemaColumn? schemaColumn,
}) async {
final (_, columnsByName) = _schema[tableName]!;
final resolvedSchemaColumn = schemaColumn ?? columnsByName[columnName];
final resolvedSchemaColumn =
schemaColumn ?? _schemaColumn(_schema, tableName, columnName);
if (resolvedSchemaColumn == null) return false;
final fieldKey = (tableName, rowId, columnName);
@ -659,7 +715,10 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
return {
for (final MapEntry(key: columnName, value: value) in data.entries)
if (columnName != 'id' && columns.containsKey(columnName)) columnName: value,
if (columnName != 'id' && columns.containsKey(columnName))
columnName: columns[columnName]!.columnType == ColumnType.uuid
? _uuidValueFromDatabase(value)
: value,
};
}
}

View file

@ -306,7 +306,6 @@ class CrdtUniqueConflictResolver {
required Map<_MergeFieldKey, CrdtDataField> fields,
required Transaction transaction,
}) async {
final (_, columnsByName) = _recorder._schema[tableName]!;
final uniqueColumnNames = {
for (final column in uniqueIndex.columns) column.columnName,
};
@ -315,7 +314,7 @@ class CrdtUniqueConflictResolver {
for (final columnName in uniqueColumnNames) {
final field = fields[(tableName, row.uuidRowId, columnName)];
if (field == null) {
final columnId = columnsByName[columnName]?.id;
final columnId = _schemaColumn(_recorder._schema, tableName, columnName)?.id;
if (columnId != null) missingColumnIds.add(columnId);
continue;
}
@ -356,8 +355,8 @@ class CrdtUniqueConflictResolver {
tableName: tableName,
predicates: [
for (final MapEntry(key: columnName, value: value) in values.entries)
'd."${_escapeIdentifier(columnName)}" = ${_sqlLiteral(value)}',
'd."id" <> ${_sqlLiteral(rowId)}',
_domainColumnPredicate(columnName, value),
_domainColumnNotPredicate('id', rowId),
],
transaction: transaction,
);

View file

@ -12,6 +12,7 @@ import 'schema.dart';
import 'session.dart';
part 'merge.dart';
part 'merge_utils/foreign_key_projector.dart';
part 'merge_utils/unique_resolver.dart';
typedef _ReferencingForeignKey = ({
@ -21,6 +22,18 @@ typedef _ReferencingForeignKey = ({
ForeignKeyAction action,
});
typedef _ForeignKeyEdge = ({
String childTableName,
String childColumn,
String parentTableName,
String parentColumn,
ForeignKeyAction action,
bool childNullable,
Object? defaultValue,
});
typedef _ForeignKeyValueKey = (String tableName, String columnName, String value);
typedef _UniqueIndexConflictRelease = ({
List<_UniqueColumnConflictRelease> columns,
});
@ -98,6 +111,45 @@ class CrdtMutationRecorder {
return result;
}();
late final List<_ForeignKeyEdge> _foreignKeyEdges = [
for (final table in _tableDefinitionsByName.values)
if (_isCrdtTrackedTableName(table.name))
for (final foreignKey in table.foreignKeys)
if (foreignKey.columns.length == 1 &&
foreignKey.referenceColumns.length == 1 &&
_isCrdtTrackedTableName(foreignKey.referenceTable))
(
childTableName: table.name,
childColumn: foreignKey.columns.single,
parentTableName: foreignKey.referenceTable,
parentColumn: foreignKey.referenceColumns.single,
action: foreignKey.onDelete ?? ForeignKeyAction.noAction,
childNullable:
_columnsByTableAndName[table.name]![foreignKey.columns.single]!
.isNullable,
defaultValue: _defaultValueForColumn(
table.name,
foreignKey.columns.single,
),
),
];
late final Map<String, List<_ForeignKeyEdge>> _foreignKeyEdgesByParentTable = {
for (final tableName in _tableDefinitionsByName.keys)
tableName: [
for (final edge in _foreignKeyEdges)
if (edge.parentTableName == tableName) edge,
],
};
late final Map<String, List<_ForeignKeyEdge>> _foreignKeyEdgesByChildTable = {
for (final tableName in _tableDefinitionsByName.keys)
tableName: [
for (final edge in _foreignKeyEdges)
if (edge.childTableName == tableName) edge,
],
};
final _uniqueIndexesByTableName = <String, List<_UniqueIndexConflictRelease>>{};
late final Map<String, Map<String, ColumnDefinition>> _columnsByTableAndName = {
@ -220,6 +272,11 @@ class CrdtMutationRecorder {
transaction: transaction,
ignoreConflicts: true,
);
await _recordForeignKeyInsertAttempts(tableName, rowIds, transaction, {});
if (_tableColumnsMayAffectForeignKeys(tableName, null)) {
await _projectForeignKeys(transaction);
}
});
}
@ -248,6 +305,15 @@ class CrdtMutationRecorder {
transaction,
);
await _recordUpdatedFields(reinsertedRows, crdtDataRows, null, transaction);
await _recordForeignKeyAttemptsForRows(
tableName,
rowIds,
null,
transaction,
);
if (_tableColumnsMayAffectForeignKeys(tableName, null)) {
await _projectForeignKeys(transaction);
}
});
}
@ -270,7 +336,33 @@ class CrdtMutationRecorder {
'updated',
transaction,
);
await _recordUpdatedFields(updatedRows, crdtDataRows, columns, transaction);
final implicitForeignKeyRepairFields = columns == null
? await _findImplicitForeignKeyRepairFields(
tableName: tableName,
rowIds: rowIds,
transaction: transaction,
)
: const <_MergeFieldKey>{};
await _recordUpdatedFields(
updatedRows,
crdtDataRows,
columns,
transaction,
skippedFields: implicitForeignKeyRepairFields,
);
await _recordForeignKeyAttemptsForRows(
tableName,
rowIds,
columns?.map((column) => column.columnName).toSet(),
transaction,
skippedFields: implicitForeignKeyRepairFields,
);
if (_tableColumnsMayAffectForeignKeys(
tableName,
columns?.map((column) => column.columnName).toSet(),
)) {
await _projectForeignKeys(transaction);
}
});
}
@ -349,11 +441,9 @@ 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"
LEFT JOIN "crdt_data_tombstone" tomb
ON tomb."rowId" = r."id"
WHERE c."id" IN ($whereRowIds)
AND c."$childColumn" IS NOT NULL
AND (p."id" IS NULL OR tomb."clFlag" % 2 = 0)
AND (p."id" IS NULL OR r."visibility" > $crdtRowLastVisibleVisibilityIndex)
''';
});
@ -451,8 +541,9 @@ WHERE c."id" IN ($whereRowIds)
List<T> updatedRows,
List<CrdtDataRow> crdtDataRows,
List<Column>? columns,
Transaction transaction,
) async {
Transaction transaction, {
Set<_MergeFieldKey> skippedFields = const {},
}) async {
if (updatedRows.isEmpty || crdtDataRows.isEmpty) return;
final crdtDataRowByUuid = {
@ -473,6 +564,26 @@ WHERE c."id" IN ($whereRowIds)
'No CRDT schema column for ${table.tableName}.${column.columnName}',
)),
];
await _upsertCrdtFieldsForRows(
table.tableName,
[
for (final row in updatedRows) crdtDataRowByUuid[row.id as UuidValue]!,
],
schemaColumns,
transaction,
skippedFields: skippedFields,
);
}
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(
@ -485,159 +596,16 @@ WHERE c."id" IN ($whereRowIds)
for (final f in existingFields) (f.rowId, f.columnId): f,
};
final hlcManager = _getHlcManager(transaction);
final toInsert = <CrdtDataField>[];
final toUpdate = <CrdtDataField>[];
for (final row in updatedRows) {
final crdtDataRow = crdtDataRowByUuid[row.id as UuidValue]!;
for (final schemaCol in schemaColumns) {
final hlc = hlcManager.increment();
final rowPk = crdtDataRow.id!;
final colPk = schemaCol.id!;
final existing = fieldByRowAndColumn[(rowPk, colPk)];
if (existing == null) {
toInsert.add(
CrdtDataField(
rowId: rowPk,
columnId: colPk,
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(
_session,
toInsert,
transaction: transaction,
);
}
if (toUpdate.isNotEmpty) {
await CrdtDataField.db.update(
_session,
toUpdate,
transaction: transaction,
);
}
}
/// Soft-deletes rows by recording a tombstone instead of removing them.
///
/// Expects a matching [CrdtDataRow] per domain row (`uuidRowId` = domain id).
Future<void> insteadOfDelete<T extends TableRow>(
List<T> deletedRows,
Transaction transaction,
) async {
if (deletedRows.isEmpty) return;
if (!isCrdtTracked<T>(deletedRows.first.table)) return;
await _softDeleteRowsByTable(
deletedRows.first.table.tableName,
deletedRows.uuidRowIds,
transaction,
null,
CrdtDataDeletedReason.userDelete,
);
}
Future<void> _softDeleteRowsByTable(
String tableName,
Set<UuidValue> rowIds,
Transaction transaction,
Set<String>? processedRowIds,
CrdtDataDeletedReason reason,
) async {
if (rowIds.isEmpty) return;
if (!_isCrdtTrackedTableName(tableName)) return;
final processing = processedRowIds ?? {};
final unprocessedRowIds = rowIds
.where((rowId) => processing.add('$tableName:$rowId'))
.toSet();
if (unprocessedRowIds.isEmpty) return;
final crdtDataRows = await _findRequiredCrdtRows(
tableName,
unprocessedRowIds,
'deleted',
transaction,
includeDeleted: true,
);
try {
final visibleCrdtRows = crdtDataRows
.where((row) => row.deleted?.isDeleted != true)
.toList();
if (visibleCrdtRows.isEmpty) return;
final visibleRowIds = visibleCrdtRows.map((row) => row.uuidRowId).toSet();
await _applyForeignKeyDeleteActions(
tableName,
visibleRowIds,
transaction,
processing,
);
await _releaseUniqueConflicts(tableName, visibleRowIds, transaction);
await _markCrdtRowsDeleted(visibleCrdtRows, true, reason, transaction);
} finally {
for (final rowId in unprocessedRowIds) {
processing.remove('$tableName:$rowId');
}
}
}
/// Records field updates by table name and UUID ids.
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;
final rowPks = crdtDataRows.map((r) => r.id!).toSet();
final columnPks = schemaColumns.map((c) => c.id!).toSet();
final existingFields = await CrdtDataField.db.find(
_session,
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 = _getHlcManager(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) {
@ -670,6 +638,114 @@ WHERE c."id" IN ($whereRowIds)
}
}
/// Soft-deletes rows by recording a tombstone instead of removing them.
///
/// Expects a matching [CrdtDataRow] per domain row (`uuidRowId` = domain id).
Future<void> insteadOfDelete<T extends TableRow>(
List<T> deletedRows,
Transaction transaction,
) async {
if (deletedRows.isEmpty) return;
if (!isCrdtTracked<T>(deletedRows.first.table)) return;
await _softDeleteRowsByTable(
deletedRows.first.table.tableName,
deletedRows.uuidRowIds,
transaction,
null,
CrdtDataDeletedReason.userDelete,
);
if (_tableColumnsMayAffectForeignKeys(
deletedRows.first.table.tableName,
null,
)) {
await _projectForeignKeys(transaction);
}
}
Future<void> _softDeleteRowsByTable(
String tableName,
Set<UuidValue> rowIds,
Transaction transaction,
Set<String>? processedRowIds,
CrdtDataDeletedReason reason,
) async {
if (rowIds.isEmpty) return;
if (!_isCrdtTrackedTableName(tableName)) return;
final processing = processedRowIds ?? {};
final unprocessedRowIds = rowIds
.where((rowId) => processing.add('$tableName:$rowId'))
.toSet();
if (unprocessedRowIds.isEmpty) return;
final crdtDataRows = await _findRequiredCrdtRows(
tableName,
unprocessedRowIds,
'deleted',
transaction,
includeDeleted: true,
);
try {
final visibleCrdtRows = crdtDataRows.where((row) => !row.isHidden).toList();
if (visibleCrdtRows.isEmpty) return;
final visibleRowIds = visibleCrdtRows.map((row) => row.uuidRowId).toSet();
final cascadeDeletes = await _applyForeignKeyDeleteActions(
tableName,
visibleRowIds,
transaction,
processing,
);
await _releaseUniqueConflicts(tableName, visibleRowIds, transaction);
await _markCrdtRowsDeleted(visibleCrdtRows, true, reason, transaction);
for (final MapEntry(key: childTableName, value: childIds)
in cascadeDeletes.entries) {
await _softDeleteRowsByTable(
childTableName,
childIds,
transaction,
processing,
CrdtDataDeletedReason.userCascadeDelete,
);
}
} finally {
for (final rowId in unprocessedRowIds) {
processing.remove('$tableName:$rowId');
}
}
}
/// Records field updates by table name and UUID ids.
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,
bool isDeleted,
@ -731,6 +807,43 @@ WHERE c."id" IN ($whereRowIds)
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(
_session,
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,
);
}
/// Whether the given row is soft-deleted.
@ -748,12 +861,7 @@ WHERE c."id" IN ($whereRowIds)
);
if (crdtRows.isEmpty) return false;
final tombstone = await CrdtDataDeleted.db.findFirstRow(
_session,
where: (t) => t.rowId.equals(crdtRows.single.id),
transaction: transaction,
);
return tombstone?.isDeleted ?? false;
return crdtRows.single.isHidden;
}
Future<List<CrdtDataRow>> _findCrdtRows(
@ -773,14 +881,15 @@ WHERE c."id" IN ($whereRowIds)
);
}
Future<void> _applyForeignKeyDeleteActions(
Future<Map<String, Set<UuidValue>>> _applyForeignKeyDeleteActions(
String parentTableName,
Set<UuidValue> parentIds,
Transaction transaction,
Set<String> processing,
) async {
if (parentIds.isEmpty) return;
if (parentIds.isEmpty) return {};
final cascadeDeletes = <String, Set<UuidValue>>{};
final foreignKeys =
_foreignKeysByReferencedTable[parentTableName] ??
const <_ReferencingForeignKey>[];
@ -850,16 +959,14 @@ WHERE c."id" IN ($whereRowIds)
transaction,
);
case ForeignKeyAction.cascade:
await _softDeleteRowsByTable(
reference.childTableName,
childIds,
transaction,
processing,
CrdtDataDeletedReason.cascadeDelete,
);
cascadeDeletes
.putIfAbsent(reference.childTableName, () => {})
.addAll(childIds);
}
}
}
return cascadeDeletes;
}
Future<void> _releaseUniqueConflicts(
@ -930,7 +1037,7 @@ WHERE c."id" IN ($whereRowIds)
}) async {
return _findVisibleDomainRowIdsWhere(
tableName: tableName,
predicates: ['d."${_escapeIdentifier(columnName)}" = ${_sqlLiteral(value)}'],
predicates: [_domainColumnPredicate(columnName, value)],
transaction: transaction,
);
}
@ -948,10 +1055,8 @@ 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"
LEFT JOIN "crdt_data_tombstone" tomb
ON tomb."rowId" = r."id"
WHERE (${predicates.join(') AND (')})
AND (tomb."id" IS NULL OR tomb."clFlag" % 2 = 1)
AND (r."id" IS NULL OR r."visibility" <= $crdtRowLastVisibleVisibilityIndex)
''',
transaction: transaction,
);
@ -1158,6 +1263,58 @@ String _sqlLiteral(Object? value) => ValueEncoder.instance.convert(value);
String _sqlLiteralList(Iterable<Object?> values) =>
values.map(ValueEncoder.instance.convert).join(', ');
String _domainColumnPredicate(
String columnName,
Object? value, {
String alias = 'd',
}) => '$alias."${_escapeIdentifier(columnName)}" = ${_sqlLiteral(value)}';
String _domainColumnNotPredicate(
String columnName,
Object? value, {
String alias = 'd',
}) => '$alias."${_escapeIdentifier(columnName)}" <> ${_sqlLiteral(value)}';
UuidValue? _uuidValueFromDatabase(Object? value) {
if (value == null) return null;
if (value is UuidValue) return value;
if (value is String) return UuidValue.withValidation(value);
return UuidValueJsonExtension.fromJson(value);
}
CrdtSchemaColumn? _schemaColumn(
Map<String, (int, Map<String, CrdtSchemaColumn>)> schema,
String tableName,
String columnName,
) => schema[tableName]?.$2[columnName];
Map<String, int> _schemaColumnIds(
Map<String, (int, Map<String, CrdtSchemaColumn>)> schema,
String tableName,
Iterable<String> columnNames,
) => {
for (final columnName in columnNames)
columnName: ?_schemaColumn(schema, tableName, columnName)?.id,
};
extension on CrdtDataDeletedReason {
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,
};
}
}
extension on List<TableRow> {
Set<UuidValue> get uuidRowIds {
if (isEmpty) return {};

View file

@ -1,16 +1,17 @@
import 'package:meta/meta.dart';
import 'package:serverpod_database/serverpod_database.dart';
import '../crdt/extensions.dart';
import '../protocol/protocol.dart';
/// Merges [where] with CRDT tombstone predicates and mutates [include] in place.
/// Merges [where] with CRDT visibility predicates and mutates [include] in place.
///
/// This method ensures that no rows marked as deleted by the CLFlag are
/// 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.
///
/// We do **not** add a root predicate for keys whose nested include is an
/// [IncludeList] - tombstone filtering for that path is already handled via the
/// [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
/// single [IncludeObject]).
@internal
@ -24,7 +25,7 @@ Expression? mergeWhereWithTombstone<T extends TableRow>(
final rootTable = serializationManager.getTableForType(T);
var merged = _mergeWhereOptional(
where,
rootTable?.whereNotDeletedOnTombstone,
rootTable?.whereVisibleOnCrdtRow,
);
merged = _mergeWhereOptional(merged, includeObjectPredicates);
return merged;
@ -41,7 +42,7 @@ Expression? _walkIncludeGraphForTombstone(
// List relation: filter inside the list subquery only.
inc.where = _mergeWhereOptional(
inc.where,
inc.table.whereNotDeletedOnTombstone,
inc.table.whereVisibleOnCrdtRow,
);
return _walkIncludeGraphForTombstone(inc.include, includeObjectPredicates);
}
@ -51,7 +52,7 @@ Expression? _walkIncludeGraphForTombstone(
for (final entry in obj.includes.entries) {
final nested = entry.value;
if (nested is IncludeList) {
// e.g. one-to-many: tombstone applies via IncludeList.where, not the main WHERE.
// e.g. one-to-many: visibility applies via IncludeList.where, not the main WHERE.
acc = _walkIncludeGraphForTombstone(nested, acc);
continue;
}
@ -61,7 +62,7 @@ Expression? _walkIncludeGraphForTombstone(
if (childTable != null) {
acc = _mergeWhereOptional(
acc,
childTable.whereNotDeletedOnTombstone,
childTable.whereVisibleOnCrdtRow,
);
}
acc = _walkIncludeGraphForTombstone(nested, acc);
@ -77,17 +78,16 @@ Expression? _mergeWhereOptional(Expression? where, Expression? addition) {
}
extension on Table {
/// Creates a predicate that filters out tombstone rows for the given [Table]
/// similarly to the generated field [CrdtDataRow.deleted].
/// Creates a predicate that filters out hidden rows for the given [Table]
/// using materialized [CrdtDataRow.visibility] metadata.
///
/// Join path:
/// - UserModel.[id] = [CrdtDataRow.uuidRowId],
/// - [CrdtDataRow.id] = [CrdtDataDeleted.rowId].
/// - UserModel.[id] = [CrdtDataRow.uuidRowId].
///
/// Returns null when [Table] does not use a UUID primary key. Keeps rows
/// when there is no tombstone row or [CrdtDataDeleted.clFlag] is odd
/// (LEFT JOIN null-safe).
Expression? get whereNotDeletedOnTombstone {
/// 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 {
if (id is! ColumnUuid) return null;
final crdtRowTable = createRelationTable<CrdtDataRowTable>(
relationFieldName: '${tableName}_crdt_row',
@ -97,7 +97,9 @@ extension on Table {
createTable: (foreignTableRelation) =>
CrdtDataRowTable(tableRelation: foreignTableRelation),
);
final tomb = crdtRowTable.deleted;
return (tomb.id.equals(null)) | Expression('${tomb.clFlag} % 2 = 1');
return (crdtRowTable.id.equals(null)) |
Expression(
'${crdtRowTable.visibility} <= $crdtRowLastVisibleVisibilityIndex',
);
}
}

View file

@ -1,10 +1,9 @@
### Why a CRDT row entered its current visibility generation.
### Why a synced user tombstone changed row visibility.
enum: CrdtDataDeletedReason
serialized: byIndex
values:
- userInsert
- userDelete
- userReinsert
- restrictRestore
- cascadeDelete
- userDelete
- userCascadeDelete
- uniqueLoser

View file

@ -16,6 +16,10 @@ fields:
### The node that updated the field.
node: CrdtNode?, relation
### The foreign key information for the field, if this field holds a
### reference to another row.
foreignKey: CrdtDataForeignKey?, relation(optional, name=field_foreign_key)
indexes:
crdt_data_fields_row_column_idx:
fields: rowId, columnId

View file

@ -0,0 +1,40 @@
### CRDT field foreign key value and projection table.
###
### This table stores the user/attempted foreign key value for a CRDT field and
### the materialized foreign key projection for that value. CrdtDataField stores
### only HLC metadata, so FK columns store their durable attempted value here.
###
### Repairs must be recomputable from merged CRDT row, field, and tombstone
### facts plus this attempted FK value. The projection columns record what was
### materialized into the domain row; they do not decide business semantics or
### create user field updates during merge.
class: CrdtDataForeignKey
table: crdt_data_foreign_key
database: all
fields:
### The field that the foreign key belongs to.
field: CrdtDataField?, relation(name=field_foreign_key, field=fieldId, onDelete=Cascade)
fieldId: int
### The user-attempted foreign key value that the CRDT field currently carries.
###
### This may be null for nullable foreign keys. This is durable FK field-value
### storage for CRDT columns whose CrdtDataField row stores only HLC metadata.
attemptedValue: UuidValue?, required
### The foreign key value that is currently visible in the actual data row
### when an override is active (overrideReason is non-null).
visibleValue: UuidValue?
### The authoritative indicator and reason for an active projection override.
###
### This is the single source of truth for whether an override is active:
### non-null means an override is active; null means no override. It also
### names the cause of the override. The FK resolver recomputes it from merged
### row, field, and tombstone facts.
overrideReason: CrdtForeignKeyOverrideReason?
indexes:
crdt_data_foreign_key_field_idx:
fields: fieldId
unique: true

View file

@ -0,0 +1,15 @@
### Why a foreign key projection override is active.
enum: CrdtForeignKeyOverrideReason
serialized: byIndex
values:
### The attempted parent target is hidden or missing and the FK projection
### set the column to null.
- setNull
### The attempted parent target is hidden or missing and the FK projection
### set the column to the default value.
- setDefault
### The attempted parent target is hidden or missing and the FK projection
### cannot repair the value (restrict, no action, cascade, or the column is
### non-nullable), so the row is hidden by projection while the attempt is
### preserved.
- missingParent

View file

@ -22,10 +22,16 @@ fields:
### The node that inserted the row with the HLC timestamp.
node: CrdtNode?, relation
### The visibility CLFlag for this row.
### Materialized visibility state for domain queries.
###
### Only present after the first visibility change event has been recorded.
### If null, the row is visible and has never been deleted/hidden.
### Updated by user tombstone writes and foreign key projection. Visible
### states are listed before hidden states in CrdtDataRowVisibility.
visibility: CrdtDataRowVisibility, default=userInsert
### The synced user tombstone for this row.
###
### Only present after the first user visibility change event has been recorded.
### Projection may keep a visible [visibility] while a user delete tombstone exists.
deleted: CrdtDataDeleted?, relation(name=row_deleted, optional)
### The fields for this row.

View file

@ -0,0 +1,13 @@
### Materialized visibility state for a CRDT row.
###
### Visible states are listed first, hidden states last. A row is visible when
### its serialized index is less than or equal to [foreignKeyRestrictRestore].
enum: CrdtDataRowVisibility
serialized: byIndex
values:
- userInsert
- userReinsert
- foreignKeyRestrictRestore
- userDelete
- userCascadeDelete
- foreignKeyCascade