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

fix: Fix the conversion between CrdtDataEntry and Insertable

Browse source
This commit is contained in:
Marcelo Soares 2026-01-25 22:52:16 -03:00
commit 825fd4d97c
No known key found for this signature in database
GPG key ID: CB2F09CE66E99EAD
10 changed files with 196 additions and 132 deletions
Download patch file Download diff file Expand all files Collapse all files

1
lib/drift_offline_first.dart
View file

@ -4,6 +4,7 @@
library;
export 'src/crdt.dart';
export 'src/crdt/extensions/convert.dart';
export 'src/crdt/extensions/entry.dart';
export 'src/crdt/extensions/extract.dart';
export 'src/crdt_base.dart';

59
lib/src/crdt/extensions/convert.dart Normal file
View file

@ -0,0 +1,59 @@
import 'package:drift/drift.dart';
import '../../database.dart';
import '../../hlc/converter.dart';
/// Extensions for [CrdtDatabase] to convert data to [CrdtDataEntry].
extension CrdtDatabaseConvertExtensions on CrdtDatabase {
/// Converts the [Insertable] object to a list of [CrdtDataEntry].
///
/// Note that the generated entries will have the same HLC timestamp for all
/// columns and will overwrite individual column updates if newer.
///
/// Returns a list of [CrdtDataEntry].
Iterable<CrdtDataEntry> convertToCrdtDataEntry<T extends Insertable>(
T data,
Hlc hlcTimestamp,
) {
final json = data.toJsonForDatabase(_generationContext);
final tableInfo = synchronizedTables.whereType<TableInfo<dynamic, T>>().single;
final rowIdValues = [for (final c in tableInfo.$primaryKey) json[c.$name]];
return tableInfo.$columns.map(
(c) => CrdtDataEntry(
userId: userId,
tblName: tableInfo.actualTableName,
columnName: c.$name,
rowId: rowIdValues.join(sqlBuilder.rowIdSeparator),
rawValue: json[c.$name]?.toDriftAny(),
hlcTimestamp: hlcTimestamp,
),
);
}
/// Gets a [GenerationContext] for the database.
GenerationContext get _generationContext => GenerationContext.fromDb(this);
}
extension on Insertable {
Map<String, Object?> toJsonForDatabase(GenerationContext context) {
final expressions = toColumns(false);
return {
for (final e in expressions.entries) e.key: e.value.extractValue(context),
};
}
}
extension on Expression {
Object? extractValue(GenerationContext context) {
final exp = this;
if (exp is! Variable) {
throw ArgumentError.value(exp, 'exp', 'Expected a Variable');
}
return exp.mapToSimpleValue(context);
}
}
extension on Object {
DriftAny? toDriftAny() => DriftAny(this);
}

51
lib/src/crdt/extensions/entry.dart
View file

@ -1,9 +1,13 @@
import 'package:drift/drift.dart';
import '../../database.dart';
import '../../hlc/converter.dart';
/// Extensions for [Iterable<CrdtDataEntry>] to get the affected tables and row IDs.
/// Extensions for the [CrdtDataEntry] class.
extension CrdtDataEntryUnwrapExtensions on CrdtDataEntry {
/// Returns the unwrapped raw value, if any.
Object? get unwrappedValue => rawValue?.rawSqlValue;
}
/// Extensions for an [Iterable] of [CrdtDataEntry].
extension CrdtDataEntryExtensions on Iterable<CrdtDataEntry> {
/// Gets the affected tables.
Iterable<String> get affectedTables => map((e) => e.tblName).toSet();
@ -14,41 +18,8 @@ extension CrdtDataEntryExtensions on Iterable<CrdtDataEntry> {
/// Gets the entries for a given table.
Iterable<CrdtDataEntry> forTable(String tableName) =>
where((e) => e.tblName == tableName);
}
/// Extensions to convert Drift [Insertable] objects to [CrdtDataEntry].
extension CrdtDataEntryConversion<T extends DataClass> on Insertable<T> {
/// Converts the [Insertable] object to a list of [CrdtDataEntry].
///
/// Note that the generated entries will have the same HLC timestamp for all
/// columns and will overwrite individual column updates if newer. The
/// [database] parameter is required to find the table information to map
/// the columns.
///
/// Returns a list of [CrdtDataEntry].
Iterable<CrdtDataEntry> toCrdtDataEntry(CrdtDatabase database, Hlc hlcTimestamp) {
final json = (this as DataClass).toJson();
final tableInfo =
database.synchronizedTables.whereType<TableInfo<dynamic, T>>().single;
final rowId = [
for (final c in tableInfo.$primaryKey) json[c.$name],
].join(database.sqlBuilder.rowIdSeparator);
return tableInfo.$columns.map(
(c) => CrdtDataEntry(
userId: database.userId,
tblName: tableInfo.actualTableName,
columnName: c.$name,
rowId: rowId,
rawValue: (json[c.$name] as Object?)?.toDriftAny(),
hlcTimestamp: hlcTimestamp,
),
);
}
}
extension on Object {
DriftAny? toDriftAny() => DriftAny(this);
/// Returns the entries with the HLC replaced by the [hlcTimestamp].
Iterable<CrdtDataEntry> withHlc(Hlc hlcTimestamp) =>
map((e) => e.copyWith(hlcTimestamp: hlcTimestamp));
}

48
lib/src/crdt/extensions/extract.dart
View file

@ -5,36 +5,44 @@ import '../../database.dart';
/// Extensions for [CrdtDatabase] to extract data from the CRDT data table.
extension CrdtDatabaseExtractExtensions on CrdtDatabase {
/// Gets a single data entry from the CRDT data table.
Future<T?> getSingleFromCrdtData<T extends Insertable>({
required String tableName,
required String rowId,
}) async {
final data = await getFromCrdtData<T>(tableName: tableName, rowIds: [rowId]);
Future<T?> getSingleFromCrdtData<T extends Insertable>(String rowId) async {
final data = await getFromCrdtData<T>([rowId]);
return data.firstOrNull;
}
/// Gets a list of data entries from the CRDT data table.
Future<List<T>> getFromCrdtData<T extends Insertable>({
required String tableName,
required Iterable<String> rowIds,
}) async {
final tableInfo = synchronizedTables.find(tableName);
final crdtDataEntries = await managers.crdtDataTable
.filter((o) => o.tblName.equals(tableName) & o.rowId.isIn(rowIds))
.get();
Future<Iterable<T>> getFromCrdtData<T extends Insertable>([
Iterable<String>? rowIds,
]) async {
final tableInfo = synchronizedTables.find<T>();
final tableName = tableInfo.actualTableName;
final crdtDataEntries = await managers.crdtDataTable.filter(
(o) {
var condition = o.tblName.equals(tableName);
if (rowIds != null) {
condition &= o.rowId.isIn(rowIds);
}
return condition;
},
).get();
return crdtDataEntries.convert<T>(tableInfo);
return tableInfo.fromCrdtDataEntries<T>(crdtDataEntries);
}
}
extension on List<CrdtDataEntry> {
Future<List<T>> convert<T extends Insertable>(TableInfo tableInfo) async {
final foundRowIds = map((e) => e.rowId).toSet();
/// Extensions for [TableInfo] to convert [CrdtDataEntry] to [Insertable].
extension ConvertCrdtDataEntry on TableInfo {
/// Converts [CrdtDataEntry] to [Insertable].
Future<Iterable<T>> fromCrdtDataEntries<T extends Insertable>(
Iterable<CrdtDataEntry> entries,
) async {
final foundRowIds = entries.map((e) => e.rowId).toSet();
return [
for (final rowId in foundRowIds)
await tableInfo.map({
for (final c in tableInfo.$columns)
c.$name: firstWhere((e) => e.rowId == rowId && e.columnName == c.$name)
await map({
for (final c in $columns)
c.$name: entries
.firstWhere((e) => e.rowId == rowId && e.columnName == c.$name)
.rawValue
?.rawSqlValue,
}) as T,

13
lib/src/database.dart
View file

@ -46,6 +46,7 @@ class CrdtDatabase extends _$CrdtDatabase {
/// should ever be run on this database to avoid messing with the user schema.
@override
int schemaVersion = 1;
}
/// Extension methods for the [List<TableInfo>] class to find a synchronized table.
extension FindSynchronizedTable on List<TableInfo> {
@ -55,11 +56,17 @@ extension FindSynchronizedTable on List<TableInfo> {
/// Gets the synchronized table information for the given table name.
///
/// If the table is not found, an [ArgumentError] is thrown.
TableInfo find(String tableName) {
TableInfo find<T extends Insertable>([String? tableName]) {
if (tableName == null && T == dynamic) {
throw ArgumentError('Table name is required when T is dynamic.');
}
return firstWhere(
(t) => t.actualTableName == tableName,
(t) =>
(tableName == null && t is TableInfo<Table, T>) ||
t.actualTableName == tableName,
orElse: () => throw ArgumentError(
'Table "$tableName" not found in synchronized tables.',
'Table ${tableName != null ? '"$tableName"' : 'for type "$T"'} not '
'found in synchronized tables.',
),
);
}

96
test/crdt/crdt_data_convert_test.dart
View file

@ -13,20 +13,7 @@ void main() {
late Hlc testHlc;
setUp(() async {
createdData = TableWithEveryColumnTypeData(
id: const RowId(1),
aBool: true,
aDateTime: DateTime.now(),
aText: 'test',
anInt: 1,
anInt64: BigInt.one,
aReal: 1.0,
aBlob: Uint8List.fromList([1, 2, 3]),
anIntEnum: TodoStatus.open,
aTextWithConverter: MyCustomObject('test'),
aUuid: UuidValue.fromString('123e4567-e89b-12d3-a456-426614174000'),
);
createdData = _createTestData();
(_, crdtDb, _) = database.crdtContext;
testHlc = Hlc.now(crdtDb.nodeId);
});
@ -35,7 +22,7 @@ void main() {
late Iterable<CrdtDataEntry> entries;
setUp(() async {
entries = createdData.toCrdtDataEntry(crdtDb, testHlc).toList();
entries = crdtDb.convertToCrdtDataEntry(createdData, testHlc);
});
test('then it creates an entry for each column.', () async {
@ -63,32 +50,69 @@ void main() {
});
test('then all entries have expected column names and values.', () async {
final entriesMap = {
for (final entry in entries) entry.columnName: entry.rawValue,
};
final mappedData =
await database.tableWithEveryColumnType.fromCrdtDataEntries(entries);
expect(mappedData.single, equals(createdData));
});
});
});
expect(entriesMap['a_bool'], equals(createdData.aBool?.driftAny));
expect(entriesMap['a_date_time'], isNotNull);
expect(entriesMap['a_text'], equals(createdData.aText?.driftAny));
expect(entriesMap['an_int'], equals(createdData.anInt?.driftAny));
expect(entriesMap['an_int64'], equals(createdData.anInt64?.driftAny));
expect(entriesMap['a_real'], equals(createdData.aReal?.driftAny));
expect(entriesMap['a_blob'], equals(createdData.aBlob?.driftAny));
group('Given a changeset converted from a data object with every column type', () {
late OfflineSyncCrdt crdt;
late CrdtDatabase crdtDb;
late TableWithEveryColumnTypeData createdData;
late List<CrdtDataEntry> expectedChangeset;
late Hlc testHlc;
setUp(() async {
createdData = _createTestData();
(crdt, crdtDb, _) = database.crdtContext;
// Perform an insert to the database to get the exact changeset that would
// be generated by the CRDT triggers.
await database.tableWithEveryColumnType.insertOne(createdData);
expectedChangeset = await crdtDb.crdtDataTable.all().get();
testHlc = expectedChangeset.first.hlcTimestamp;
await crdtDb.crdtDataTable.deleteAll();
});
group('when saving the changeset to the CRDT database', () {
late Iterable<CrdtDataEntry> entries;
setUp(() async {
entries = crdtDb.convertToCrdtDataEntry(createdData, testHlc);
await crdt.saveChangeset(entries, [testHlc]);
});
test('then the saved data is equal to the generated by CRDT triggers.', () async {
final savedChangeset = await crdtDb.managers.crdtDataTable.get();
// The list object is not equal to the other objects because it is a
// different instance. So we remove it from the list and compare the
// it separately.
final savedListObject = savedChangeset.removeAt(7);
final expectedListObject = expectedChangeset.removeAt(7);
expect(savedChangeset, equals(expectedChangeset));
expect(
entriesMap['an_int_enum'],
equals(createdData.anIntEnum?.index.driftAny),
savedListObject.unwrappedValue,
equals(expectedListObject.unwrappedValue),
);
expect(
// aTextWithConverter uses 'insert' as column name
entriesMap['insert']?.rawSqlValue,
equals(const DriftAny({'data': 'test'}).rawSqlValue),
);
expect(entriesMap['a_uuid'], createdData.aUuid?.driftAny);
});
});
});
}
extension on Object {
DriftAny? get driftAny => DriftAny(this);
}
TableWithEveryColumnTypeData _createTestData() => TableWithEveryColumnTypeData(
id: const RowId(1),
aBool: true,
aDateTime: DateTime.now(),
aText: 'test',
anInt: 1,
anInt64: BigInt.one,
aReal: 1.0,
aBlob: Uint8List.fromList([1, 2, 3]),
anIntEnum: TodoStatus.open,
aTextWithConverter: MyCustomObject('test'),
aUuid: UuidValue.fromString('123e4567-e89b-12d3-a456-426614174000'),
);

20
test/crdt/crdt_data_extract_test.dart
View file

@ -37,8 +37,7 @@ void main() {
'when getting the data from the CRDT data table of the CRDT database '
'then it rebuilds the data perfectly.', () async {
final data = await crdtDb.getSingleFromCrdtData<TableWithEveryColumnTypeData>(
tableName: 'table_with_every_column_type',
rowId: rowId,
rowId,
);
expect(data, equals(createdData));
@ -48,8 +47,7 @@ void main() {
'when trying to get the data from the CRDT data table with a non-existent row id '
'then it returns null.', () async {
final data = await crdtDb.getSingleFromCrdtData<TableWithEveryColumnTypeData>(
tableName: 'table_with_every_column_type',
rowId: 'non-existent-row-id',
'non-existent-row-id',
);
expect(data, isNull);
@ -59,8 +57,7 @@ void main() {
'when trying to get the data from the CRDT data table with multiple row ids '
'then it returns the data for the existing row ids.', () async {
final data = await crdtDb.getFromCrdtData<TableWithEveryColumnTypeData>(
tableName: 'table_with_every_column_type',
rowIds: [rowId, 'non-existent-row-id'],
[rowId, 'non-existent-row-id'],
);
expect(data, equals([createdData]));
@ -70,11 +67,14 @@ void main() {
'when trying to get the data from the CRDT data table for a non-synchronized table '
'then it throws an error.', () async {
expect(
() => crdtDb.getFromCrdtData<TableWithEveryColumnTypeData>(
tableName: 'table_without_pk',
rowIds: [rowId],
() => crdtDb.getSingleFromCrdtData<WithCustomTypeData>(rowId),
throwsA(
isA<ArgumentError>().having(
(e) => e.message,
'message',
'Table for type "WithCustomTypeData" not found in synchronized tables.',
),
),
throwsA(isA<ArgumentError>()),
);
});
});

12
test/crdt/crdt_merge_changeset_test.dart
View file

@ -22,7 +22,7 @@ void main() {
targetDate: DateTime.now(),
);
changeset = todo.toCrdtDataEntry(crdtDb, Hlc.now(nodeId));
changeset = crdtDb.convertToCrdtDataEntry(todo, Hlc.now(nodeId));
});
test(
@ -49,10 +49,10 @@ void main() {
targetDate: DateTime.now(),
);
firstChangeset = todo.toCrdtDataEntry(crdtDb, Hlc.now(nodeId));
firstChangeset = crdtDb.convertToCrdtDataEntry(todo, Hlc.now(nodeId));
await crdt.merge(firstChangeset);
lateOldChangeset = todo.toCrdtDataEntry(crdtDb, Hlc.zero(nodeId));
lateOldChangeset = crdtDb.convertToCrdtDataEntry(todo, Hlc.zero(nodeId));
});
test(
@ -79,11 +79,11 @@ void main() {
targetDate: DateTime.now(),
);
firstChangeset = todo.toCrdtDataEntry(crdtDb, Hlc.now(nodeId));
firstChangeset = crdtDb.convertToCrdtDataEntry(todo, Hlc.now(nodeId));
await crdt.merge(firstChangeset);
newerChangeset = todo
.toCrdtDataEntry(crdtDb, Hlc.now(nodeId))
newerChangeset = crdtDb
.convertToCrdtDataEntry(todo, Hlc.now(nodeId))
.where((e) => e.columnName == 'content');
});

12
test/crdt/crdt_save_changeset_test.dart
View file

@ -25,7 +25,7 @@ void main() {
targetDate: DateTime.now(),
);
changeset = todo.toCrdtDataEntry(crdtDb, Hlc.now(otherNodeId));
changeset = crdtDb.convertToCrdtDataEntry(todo, Hlc.now(otherNodeId));
lastHlc = Hlc.now(otherNodeId);
});
@ -65,11 +65,11 @@ void main() {
);
lastHlc = Hlc.now(otherNodeId);
firstChangeset = todo.toCrdtDataEntry(crdtDb, lastHlc);
firstChangeset = crdtDb.convertToCrdtDataEntry(todo, lastHlc);
await crdt.saveChangeset(firstChangeset, [lastHlc]);
lateOldHlc = Hlc.zero(otherNodeId);
lateOldChangeset = todo.toCrdtDataEntry(crdtDb, lateOldHlc);
lateOldChangeset = crdtDb.convertToCrdtDataEntry(todo, lateOldHlc);
});
group('when saving a changeset with older HLC', () {
@ -108,12 +108,12 @@ void main() {
);
lastHlc = Hlc.now(otherNodeId);
firstChangeset = todo.toCrdtDataEntry(crdtDb, lastHlc);
firstChangeset = crdtDb.convertToCrdtDataEntry(todo, lastHlc);
await crdt.saveChangeset(firstChangeset, [lastHlc]);
newerHlc = Hlc.now(otherNodeId);
newerChangeset = todo
.toCrdtDataEntry(crdtDb, newerHlc)
newerChangeset = crdtDb
.convertToCrdtDataEntry(todo, newerHlc)
.where((e) => e.columnName == 'content');
});

16
test/triggers/trigger_execution_test.dart
View file

@ -56,14 +56,11 @@ void main() {
test('has the raw values inserted correctly.', () async {
final crdtData = allCrdtDataEntries.getField(rowId, 'content');
expect(
crdtData.rawValue?.rawSqlValue,
targetContent,
);
expect(crdtData.unwrappedValue, targetContent);
final targetDateCrdtData = allCrdtDataEntries.getField(rowId, 'target_date');
expect(
targetDateCrdtData.rawValue?.rawSqlValue,
targetDateCrdtData.unwrappedValue,
targetDate.toIso8600StringWithOffset(),
);
});
@ -132,14 +129,11 @@ void main() {
test('has the raw values updated correctly.', () async {
final crdtData = allCrdtDataEntries.getField(rowId, 'content');
expect(
crdtData.rawValue?.rawSqlValue,
targetContent,
);
expect(crdtData.unwrappedValue, targetContent);
final targetDateCrdtData = allCrdtDataEntries.getField(rowId, 'target_date');
expect(
targetDateCrdtData.rawValue?.rawSqlValue,
targetDateCrdtData.unwrappedValue,
targetDate.toIso8600StringWithOffset(),
);
});
@ -222,7 +216,7 @@ void main() {
test('has the deleted flag column set to true.', () async {
final crdtData = allCrdtDataEntries.getField(rowId, '__crdt_is_deleted');
expect(crdtData.rawValue?.rawSqlValue, 1);
expect(crdtData.unwrappedValue, 1);
});
test('has the deleted HLC timestamp greater than the created.', () async {