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

feat: Support specifying the conflicting bias

Browse source
This commit is contained in:
Marcelo Soares 2026-01-27 15:31:55 -03:00
commit abc16d1895
No known key found for this signature in database
GPG key ID: CB2F09CE66E99EAD
6 changed files with 205 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

28
lib/src/database/triggers.dart
View file

@ -1,5 +1,6 @@
import 'package:drift/drift.dart';
import '../migrator.dart';
import '../utils/sql_builder.dart';
/// The name of the HLC function used in the database.
@ -15,7 +16,13 @@ enum Operation { insert, update, delete }
/// Provides the CRDT control trigger statements for the database.
abstract class OfflineSyncTriggers extends CrdtDataSqlBuilder {
/// Creates a new instance of [OfflineSyncTriggers].
OfflineSyncTriggers(super.crdtDb);
OfflineSyncTriggers(
super.crdtDb, [
this.conflictingBias = ConflictingBias.deleteWins,
]);
/// The bias to use when conflicting operations are detected.
final ConflictingBias conflictingBias;
/// Wraps the provided SQL statement as a CREATE TRIGGER statement.
///
@ -52,14 +59,23 @@ abstract class OfflineSyncTriggers extends CrdtDataSqlBuilder {
tableAlias: operation == Operation.insert ? 'NEW' : 'OLD',
);
final columnNames = operation == Operation.delete
? [isDeletedColumnName]
: table.nonGenerated;
final columnNames = [
if (operation != Operation.delete) ...table.nonGenerated,
if (operation != Operation.update ||
conflictingBias == ConflictingBias.updateWins)
isDeletedColumnName,
];
final columnInserts = columnNames.map((c) {
final columnValue = c == isDeletedColumnName
? operation == Operation.delete
? 'TRUE'
: 'FALSE'
: 'NEW."$c"';
return ' SELECT\n${[
' \'$c\' AS "column_name"',
' ${c == isDeletedColumnName ? 'TRUE' : 'NEW."$c"'} AS "raw_value"',
' $columnValue AS "raw_value"',
if (operation == Operation.update) ' NEW."$c" IS NOT OLD."$c" AS "value_changed"',
].join(',\n')}';
});
@ -111,7 +127,7 @@ ${columnInserts.join('\n UNION ALL\n')}
/// Provides the CRDT control trigger statements for the SQLite3 database.
class Sqlite3OfflineSyncTriggers extends OfflineSyncTriggers {
/// Creates a new instance of [Sqlite3OfflineSyncTriggers].
Sqlite3OfflineSyncTriggers(super.crdtDb);
Sqlite3OfflineSyncTriggers(super.crdtDb, [super.conflictingBias]);
@override
String wrapAsCreateTriggerStatement({

17
lib/src/migrator.dart
View file

@ -4,6 +4,15 @@ import 'crdt/crdt.dart';
import 'database/database.dart';
import 'database/triggers.dart';
/// The bias to use when conflicting operations are detected.
enum ConflictingBias {
/// Newer updates are overridden by deletes.
deleteWins,
/// Deletes are undone by newer conflicting updates.
updateWins,
}
/// Runs migrations declared by a [MigrationStrategy].
class OfflineSyncMigrator extends Migrator {
/// Used internally by drift when opening the database.
@ -11,6 +20,7 @@ class OfflineSyncMigrator extends Migrator {
super.database, {
required this.userId,
required this.nodeId,
this.conflictingBias = ConflictingBias.deleteWins,
/// Restrict tables to synchronize. If not provided, all tables will be considered.
List<TableInfo>? synchronizedTables,
@ -38,6 +48,11 @@ class OfflineSyncMigrator extends Migrator {
/// Tables to be synchronized with CRDT.
final List<TableInfo> synchronizedTables;
/// The bias to use when conflicting operations are detected.
///
/// The default is [ConflictingBias.deleteWins], which is the most common case.
final ConflictingBias conflictingBias;
/// The database instance to apply the CRDT schema to.
late final crdtDb = CrdtDatabase(
database.executor,
@ -103,7 +118,7 @@ class OfflineSyncMigrator extends Migrator {
Future<void> _createTriggers(TableInfo table) async {
final triggerCreator = switch (database.executor.dialect) {
SqlDialect.sqlite => Sqlite3OfflineSyncTriggers(crdtDb),
SqlDialect.sqlite => Sqlite3OfflineSyncTriggers(crdtDb, conflictingBias),
// TODO: Implement the triggers for the Postgres dialect.
_ => throw UnsupportedError('Unsupported dialect: ${database.executor.dialect}'),
};

6
test/crdt/crdt_data_convert_test.dart
View file

@ -72,7 +72,11 @@ void main() {
// 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();
expectedChangeset = await crdtDb.managers.crdtDataTable
.filter(
(f) => f.columnName.equals(crdtDb.sqlBuilder.isDeletedColumnName).not(),
)
.get();
testHlc = expectedChangeset.first.hlcTimestamp;
await crdtDb.crdtDataTable.deleteAll();
});

5
test/crdt/crdt_get_changeset_test.dart
View file

@ -59,7 +59,10 @@ void main() {
() async {
final changeset = await crdt.getChangeset();
final changesetColumns = changeset.map((e) => e.columnName).toSet();
final expectedColumns = database.todosTable.$columns.map((c) => c.$name);
final expectedColumns = [
...database.todosTable.$columns.map((c) => c.$name),
crdtDb.sqlBuilder.isDeletedColumnName,
];
expect(changeset.length, expectedColumns.length);
expect(changesetColumns, equals(expectedColumns));

105
test/triggers/trigger_execution_test.dart
View file

@ -33,17 +33,22 @@ void main() {
});
group('then the trigger is executed and the CRDT data table', () {
late String isDeletedColumnName;
late List<CrdtDataEntry> allCrdtDataEntries;
setUp(() async {
final migrator = database.createMigrator();
allCrdtDataEntries = await migrator.crdtDb.managers.crdtDataTable.get();
isDeletedColumnName = migrator.crdtDb.sqlBuilder.isDeletedColumnName;
});
test('has one entry for each column of the target table.', () async {
expect(
allCrdtDataEntries.map((e) => e.columnName).toSet(),
equals(database.todosTable.$columns.map((c) => c.$name).toSet()),
equals(
database.todosTable.$columns.map((c) => c.$name).toSet()
..add(isDeletedColumnName),
),
);
});
@ -81,11 +86,11 @@ void main() {
createdRowId = await database.managers.todosTable.create(
(t) => t(content: initialContent),
);
rowId = createdRowId.toString();
final migrator = database.createMigrator();
final allCrdtDataEntries = await migrator.crdtDb.managers.crdtDataTable.get();
createdHlc = allCrdtDataEntries.first.hlcTimestamp;
rowId = createdRowId.toString();
});
group('when updating the existing row with a different value', () {
@ -206,24 +211,23 @@ void main() {
'then the trigger is executed and the deleted row in the CRDT data table',
() {
late List<CrdtDataEntry> allCrdtDataEntries;
late String rowId;
late String isDeletedColumnName;
setUp(() async {
final migrator = database.createMigrator();
allCrdtDataEntries = await migrator.crdtDb.managers.crdtDataTable.get();
rowId = createdRowId.toString();
isDeletedColumnName = migrator.crdtDb.sqlBuilder.isDeletedColumnName;
});
test('has the deleted flag column set to true.', () async {
final crdtData = allCrdtDataEntries.getField(rowId, '__crdt_is_deleted');
final crdtData = allCrdtDataEntries.getField(rowId, isDeletedColumnName);
expect(crdtData.unwrappedValue, 1);
});
test('has the deleted HLC timestamp greater than the created.', () async {
final deletedEntry = allCrdtDataEntries.getField(
rowId,
'__crdt_is_deleted',
isDeletedColumnName,
);
expect(deletedEntry.hlcTimestamp, greaterThan(createdHlc));
});
@ -231,6 +235,93 @@ void main() {
);
});
});
group('Given a database with a previously deleted row from a synchronized table', () {
const initialContent = 'test';
late String isDeletedColumnName;
late int createdRowId;
late Hlc deletedHlc;
late String rowId;
setUp(() async {
createdRowId = await database.managers.todosTable.create(
(t) => t(content: initialContent),
);
rowId = createdRowId.toString();
final migrator = database.createMigrator();
final deleted = await database.managers.todosTable
.filter((t) => t.id.equals(RowId(createdRowId)))
.delete();
expect(deleted, equals(1));
final allCrdtDataEntries = await migrator.crdtDb.managers.crdtDataTable.get();
isDeletedColumnName = migrator.crdtDb.sqlBuilder.isDeletedColumnName;
deletedHlc = allCrdtDataEntries.getField(rowId, isDeletedColumnName).hlcTimestamp;
});
group('when re-inserting the previously deleted row', () {
setUp(() async {
await database.managers.todosTable.create(
(t) => t(
id: Value(RowId(createdRowId)),
content: initialContent,
),
);
});
test(
'then the row has been re-inserted in the target table with the same data.',
() async {
final row = await database.managers.todosTable.getSingle();
expect(row, isNotNull);
expect(row.id, equals(createdRowId));
expect(row.content, initialContent);
},
);
group(
'then the trigger is executed and the re-inserted row in the CRDT data table',
() {
late List<CrdtDataEntry> allCrdtDataEntries;
setUp(() async {
final migrator = database.createMigrator();
allCrdtDataEntries = await migrator.crdtDb.managers.crdtDataTable.get();
isDeletedColumnName = migrator.crdtDb.sqlBuilder.isDeletedColumnName;
});
test('has the deleted flag column set to false.', () async {
final crdtData = allCrdtDataEntries.getField(rowId, isDeletedColumnName);
expect(crdtData.unwrappedValue, 0);
});
test(
'has the deleted flag HLC timestamp greater than the deleted.',
() async {
final deletedEntry = allCrdtDataEntries.getField(
rowId,
isDeletedColumnName,
);
expect(deletedEntry.hlcTimestamp, greaterThan(deletedHlc));
},
);
test(
'has the HLC timestamp updated for newly inserted columns.',
() async {
expect(
allCrdtDataEntries.map((e) => e.hlcTimestamp),
everyElement(greaterThan(deletedHlc)),
);
},
);
},
);
});
});
}
extension on Iterable<CrdtDataEntry> {

60
test/triggers/trigger_generation_test.dart
View file

@ -74,6 +74,10 @@ BEGIN
SELECT
'status' AS "column_name",
NEW."status" AS "raw_value"
UNION ALL
SELECT
'__crdt_is_deleted' AS "column_name",
FALSE AS "raw_value"
)
WHERE TRUE
ON CONFLICT ("user_id", "table_name", "column_name", "row_id")
@ -312,4 +316,60 @@ END;''',
});
});
});
// This test assures that the above tests are for [ConflictingBias.deleteWins]
// and avoids the need for duplicating all tests.
test('Given a SQLite3 migrator with no specified conflicting bias '
'then it generates the triggers with the deleteWins conflicting bias.', () {
final migrator = database.createMigrator();
expect(migrator.conflictingBias, ConflictingBias.deleteWins);
});
group('Given a SQLite3 database with a table and updateWins conflicting bias', () {
late OfflineSyncMigrator migrator;
late String isDeletedColumnName;
setUp(() {
migrator = OfflineSyncMigrator(
database,
userId: testUserId,
nodeId: testNodeId,
conflictingBias: ConflictingBias.updateWins,
synchronizedTables: [
database.todosTable,
],
);
isDeletedColumnName = migrator.crdtDb.sqlBuilder.isDeletedColumnName;
});
group('when generating trigger statements', () {
late List<String> triggers;
setUp(() {
final generator = Sqlite3OfflineSyncTriggers(
migrator.crdtDb,
ConflictingBias.updateWins,
);
triggers = generator.generateCreateTriggerStatements(database.todosTable);
});
test('then the update trigger includes the deleted flag column.', () {
final updateStatement = triggers.firstWhere(
(trigger) => trigger.contains('AFTER UPDATE ON "todos"'),
);
expect(
updateStatement,
contains('''
UNION ALL
SELECT
'$isDeletedColumnName' AS "column_name",
FALSE AS "raw_value",
NEW."__crdt_is_deleted" IS NOT OLD."__crdt_is_deleted" AS "value_changed"
'''),
);
});
});
});
}