feat: Basic implementation of the OfflineSyncCrdt methods on SQLite3

lib/drift_offline_first.dart

@@ -3,6 +3,8 @@
 /// More docs go here.
 library;
 
+export 'src/crdt.dart';
+export 'src/crdt_base.dart';
 export 'src/database.dart';
 export 'src/hlc/converter.dart';
 export 'src/migrator.dart';

lib/src/changeset.dart

@@ -1,42 +0,0 @@
-import 'package:crdt/crdt.dart';
-
-import 'database.dart';
-
-/// Extension method to provide conversion utilities to [CrdtDataEntry] lists.
-extension CrdtDataEntryListToChangesetExtension on List<CrdtDataEntry> {
-  /// Converts a list of [CrdtDataEntry] to a [CrdtChangeset].
-  CrdtChangeset toChangeset() {
-    final changeset = CrdtChangeset();
-    forEach(changeset.add);
-    return changeset;
-  }
-}
-
-/// Extension method to provide conversion utilities to [CrdtChangeset] objects.
-extension CrdtChangesetExtension on CrdtChangeset {
-  /// Adds a [CrdtDataEntry] to the [CrdtChangeset].
-  void add(CrdtDataEntry entry) {
-    final record = entry.toChangesetRecord();
-    if (containsKey(entry.tblName)) {
-      this[entry.tblName]!.add(record);
-    } else {
-      this[entry.tblName] = [record];
-    }
-  }
-
-  /// Converts a [CrdtChangeset] to a list of [CrdtDataEntry].
-  List<CrdtDataEntry> toCrdtDataEntries() => [
-        for (final tableEntries in entries)
-          for (final rowColumnValue in tableEntries.value)
-            CrdtDataEntry.fromJson(rowColumnValue),
-      ];
-}
-
-/// Extension method to provide conversion utilities to [CrdtDataEntry] objects.
-extension CrdtDataEntryExtension on CrdtDataEntry {
-  /// Returns the primary key for the [CrdtDataEntry] to be used in a [CrdtChangeset].
-  String get changesetPrimaryKey => '${tblName}_${columnName}_$rowId';
-
-  /// Converts a [CrdtDataEntry] to a record to be used in a [CrdtChangeset].
-  Map<String, dynamic> toChangesetRecord() => {changesetPrimaryKey: toJson()};
-}

lib/src/crdt.dart

@@ -1,118 +1,135 @@
 import 'package:crdt/crdt.dart';
 import 'package:drift/drift.dart';
 
-import 'changeset.dart';
+import 'crdt_base.dart';
 import 'database.dart';
 
 /// A CRDT implementation for offline-first synchronization.
-class OfflineSyncCrdt extends Crdt {
+class OfflineSyncCrdt extends CrdtBase {
-  /// Make sure you run [initialize] after instantiation.
+  /// Creates a new instance of [OfflineSyncCrdt].
-  OfflineSyncCrdt(this._db);
+  ///
+  /// The [nodeId] must be unique across the system.
+  OfflineSyncCrdt(this._db, {required super.userId, required super.nodeId});
 
   final CrdtDatabase _db;
 
-  /// Initialize this CRDT.
-  ///
-  /// The [nodeId] must be unique across the system. Note that setting the node
-  /// id on [initialize] only works for empty CRDTs. For non-empty CRDTs, the
-  /// node id will be read from the database.
-  Future<void> initialize({required String nodeId}) async {
-    canonicalTime = await getLastModified(onlyNodeId: nodeId);
-  }
-
   @override
-  Future<Hlc> getLastModified({String? onlyNodeId, String? exceptNodeId}) async {
+  // TODO: Remove this method lately if it is not needed - as it seems it will.
-    assert(
+  Future<Hlc?> getLastModified({String? onlyNodeId, String? exceptNodeId}) async {
-      onlyNodeId == null || exceptNodeId == null,
+    if (onlyNodeId != null && exceptNodeId != null) {
-      'Only one of onlyNodeId or exceptNodeId can be provided.',
+      throw ArgumentError('Only one of onlyNodeId or exceptNodeId can be provided.');
-    );
+    }
 
     // NOTE: Although `max` has a better performance for the general use case,
     // when the column is indexed, `order by` with `limit 1` will be just as fast.
     // The plus of using the order by approach is the native deserialization of
     // the Hlc object by Drift.
     final rowWithMaxLastSyncHlc = await _db.managers.crdtDataTable
+        .filter((o) {
+          if (onlyNodeId != null) {
+            return o.hlcTimestamp.column.contains(onlyNodeId);
+          }
+          if (exceptNodeId != null) {
+            return o.hlcTimestamp.column.contains(exceptNodeId).not();
+          }
+          return const Constant(true);
+        })
         .orderBy((o) => o.hlcTimestamp.desc())
         .limit(1)
         .getSingleOrNull();
 
-    final lastModified = rowWithMaxLastSyncHlc?.hlcTimestamp;
+    return rowWithMaxLastSyncHlc?.hlcTimestamp;
-    return lastModified ?? Hlc.zero(nodeId);
   }
 
   @override
-  Future<CrdtChangeset> getChangeset({
+  Future<Iterable<CrdtDataEntry>> getChangeset({
-    Map<String, Query>? customQueries,
-    Iterable<String>? onlyTables,
     String? onlyNodeId,
     String? exceptNodeId,
-    Hlc? modifiedOn,
     Hlc? modifiedAfter,
   }) async {
     assert(onlyNodeId == null || exceptNodeId == null);
-    assert(modifiedOn == null || modifiedAfter == null);
-
-    if (customQueries != null) {
-      throw UnimplementedError('Not implemented');
-    }
 
     // Modified times use the local node id
-    modifiedOn = modifiedOn?.apply(nodeId: nodeId);
     modifiedAfter = modifiedAfter?.apply(nodeId: nodeId);
 
     final query = _db.managers.crdtDataTable.filter((o) {
       Expression<bool> condition = const Constant(true);
 
-      if (onlyTables != null) {
-        condition &= o.tblName.isIn(onlyTables);
-      }
       if (onlyNodeId != null) {
         condition &= o.hlcTimestamp.column.contains(onlyNodeId);
       }
       if (exceptNodeId != null) {
         condition &= o.hlcTimestamp.column.contains(exceptNodeId).not();
       }
-      if (modifiedOn != null) {
-        condition &= o.hlcTimestamp.equals(modifiedOn);
-      }
       if (modifiedAfter != null) {
         condition &= o.hlcTimestamp.column.isBiggerThanValue(modifiedAfter.toString());
       }
       return condition;
     });
 
-    final rows = await query.get();
+    return query.get();
-    return rows.toChangeset();
   }
 
   @override
-  Future<void> merge(CrdtChangeset changeset) async {
+  Future<void> saveChangeset(
-    if (changeset.recordCount == 0) return;
+    Iterable<CrdtDataEntry> changeset,
-
+    Iterable<Hlc> receivedHlcs,
-    // Ignore empty records
+  ) async {
-    changeset.removeWhere((_, records) => records.isEmpty);
-
-    // Validate changeset and get new canonical time
-    final hlc = validateChangeset(changeset);
-
-    // Merge records
     await _db.transaction(() async {
-      await _db.managers.crdtDataTable.bulkCreate(
+      await _saveChangeset(changeset);
-        (_) => changeset.toCrdtDataEntries().map((e) => e.toCompanion(false)),
+      await _applyCompensationRules();
-        mode: InsertMode.insertOrReplace,
+      await _saveMergedHlcs(receivedHlcs);
-      );
-
-      await applyCompensationRules();
-      onDatasetChanged(changeset.keys, hlc);
     });
   }
 
+  /// Save the [changeset] to the database.
+  ///
+  /// Should be called within a transaction. Must ensure that only newer values
+  /// are saved to the database.
+  Future<void> _saveChangeset(Iterable<CrdtDataEntry> changeset) async {
+    return _db.managers.crdtDataTable.bulkCreate(
+      (_) => changeset.map((e) => e.toCompanion(false)),
+      mode: InsertMode.insertOrReplace,
+      onConflict: DoUpdate.withExcluded(
+        (old, excluded) => CrdtDataTableCompanion.custom(
+          rawValue: excluded.rawValue,
+          hlcTimestamp: excluded.hlcTimestamp,
+        ),
+        where: (old, excluded) => old.hlcTimestamp.isSmallerThan(excluded.hlcTimestamp),
+      ),
+    );
+  }
+
   /// Applies compensation rules to the CRDT data table to preserve invariants.
   ///
   /// If the any invariant fails to be preserved, a flag will be set on the
   /// control table to prevent further merges until the issue is resolved by
   /// the server and a repair merge is pushed to this node.
-  Future<void> applyCompensationRules() async {
+  Future<void> _applyCompensationRules() async {
     // TODO: Implement this.
   }
+
+  /// Save the [mergedHlcs] to the database.
+  ///
+  /// Should be called within a transaction after the changeset has been saved
+  /// and properly compensated.
+  Future<void> _saveMergedHlcs(Iterable<Hlc> mergedHlcs) async {
+    return _db.managers.crdtMergeHlcTable.bulkCreate(
+      (t) => mergedHlcs.map(
+        (e) => t(
+          userId: userId,
+          nodeId: e.nodeId,
+          lastReceivedHlc: e,
+        ),
+      ),
+      mode: InsertMode.insertOrReplace,
+      onConflict: DoUpdate.withExcluded(
+        (old, excluded) => CrdtMergeHlcTableCompanion.custom(
+          lastReceivedHlc: excluded.lastReceivedHlc,
+        ),
+        where: (old, excluded) =>
+            old.lastReceivedHlc.isSmallerThan(excluded.lastReceivedHlc),
+      ),
+    );
+  }
 }

lib/src/crdt_base.dart

@@ -0,0 +1,106 @@
+import 'dart:async';
+
+import 'package:crdt/crdt.dart';
+import 'package:meta/meta.dart';
+
+import 'database.dart';
+import 'hlc/stateful.dart';
+
+/// Base class for CRDT implementations.
+///
+/// This class provides the basic functionality for a CRDT implementation.
+/// It is not meant to be used directly, but rather to be extended by concrete
+/// implementations.
+abstract class CrdtBase {
+  /// Creates a new instance of [CrdtBase].
+  CrdtBase({required this.userId, required this.nodeId});
+
+  /// The user ID for the CRDT system.
+  final String userId;
+
+  /// The node ID of the CRDT.
+  final String nodeId;
+
+  /// The stateful canonical time of the CRDT.
+  late final StatefulHlc _statefulCanonicalTime = StatefulHlc.cached(userId, nodeId);
+
+  /// The canonical time of the CRDT.
+  Hlc get canonicalTime => _statefulCanonicalTime.lastHlc;
+
+  /// Returns the last modified timestamp, optionally filtering for or against a
+  /// specific node id. Useful to get "modified since" timestamps for executing
+  /// synchronization.
+  Future<Hlc?> getLastModified({
+    String? onlyNodeId,
+    String? exceptNodeId,
+  });
+
+  /// Get a list of [CrdtDataEntry] with the changes to be merged.
+  ///
+  /// Set the filtering parameters to to generate subsets:
+  /// [onlyNodeId] only records set by the specified node.
+  /// [exceptNodeId] only records not set by the specified node.
+  /// [modifiedAfter] records modified after the specified [Hlc].
+  Future<Iterable<CrdtDataEntry>> getChangeset({
+    String? onlyNodeId,
+    String? exceptNodeId,
+    Hlc? modifiedAfter,
+  });
+
+  /// Save the [changeset] to the database.
+  ///
+  /// When saving, must be careful to only override values that are older than
+  /// the [Hlc] of the changeset entry. If there are any guardrails to be
+  /// applied to the saved data, they should be applied here.
+  ///
+  /// The [receivedHlcs] is a list of HLCs, one for each external node. It must
+  /// be saved separately since it will be used to request updates from external
+  /// nodes on next syncs.
+  ///
+  /// It is recommended to use a transaction to ensure the integrity of the
+  /// data and allow for rollback in case of errors.
+  ///
+  /// Note that this method is only meant to be used internally by the [merge]
+  /// method and should not be called directly.
+  @protected
+  Future<void> saveChangeset(
+    Iterable<CrdtDataEntry> changeset,
+    Iterable<Hlc> receivedHlcs,
+  );
+
+  /// Merge [changeset] with the local dataset.
+  Future<void> merge(Iterable<CrdtDataEntry> changeset) async {
+    if (changeset.isEmpty) return;
+
+    final receivedHlcs = _extractLastReceivedHlcs(changeset);
+    await saveChangeset(changeset, receivedHlcs);
+
+    if (receivedHlcs.isEmpty) return;
+    _statefulCanonicalTime.merge(receivedHlcs.max);
+  }
+
+  /// Extracts the last received HLCs from the changeset.
+  Iterable<Hlc> _extractLastReceivedHlcs(Iterable<CrdtDataEntry> changeset) {
+    final receivedHlcs = <String, Hlc>{};
+    for (final entry in changeset) {
+      if (entry.hlcTimestamp.nodeId == nodeId) continue;
+      receivedHlcs.addOrUpdate(entry.hlcTimestamp);
+    }
+    return receivedHlcs.values;
+  }
+}
+
+extension on Iterable<Hlc> {
+  Hlc get max => reduce((a, b) => a > b ? a : b);
+}
+
+extension on Map<String, Hlc> {
+  void addOrUpdate(Hlc hlc) {
+    final registeredHlc = this[hlc.nodeId];
+    if (registeredHlc == null) {
+      this[hlc.nodeId] = hlc;
+    } else if (registeredHlc < hlc) {
+      this[hlc.nodeId] = hlc;
+    }
+  }
+}

lib/src/database.dart

@@ -3,6 +3,7 @@ import 'package:drift/drift.dart';
 import 'tables/compensation.dart';
 import 'tables/control.dart';
 import 'tables/data.dart';
+import 'tables/merge.dart';
 
 part 'database.g.dart';
 
@@ -12,11 +13,15 @@ part 'database.g.dart';
     CrdtDataTable,
     CrdtControlTable,
     CrdtCompensationTable,
+    CrdtMergeHlcTable,
   ],
 )
 class CrdtDatabase extends _$CrdtDatabase {
   /// Creates a new instance of [CrdtDatabase] with the provided [executor].
-  CrdtDatabase(super.e);
+  CrdtDatabase(super.e, {required this.synchronizedTables});
+
+  /// Tables to be synchronized with CRDT.
+  final List<TableInfo> synchronizedTables;
 
   /// Whether the database is using the SQLite or Postgres dialect.
   bool get isPostgres => executor.dialect == SqlDialect.postgres;
@@ -27,3 +32,41 @@ class CrdtDatabase extends _$CrdtDatabase {
   @override
   int schemaVersion = 1;
 }
+
+/// Extensions for [CrdtDatabase] to extract data from the CRDT data table.
+extension CrdtDatabaseDataEntryExtensions on CrdtDatabase {
+  /// Gets a single data entry from the CRDT data table.
+  Future<R?> getSingleFromCrdtData<T extends Table, R extends DataClass>(
+    String rowId,
+  ) async {
+    return (await getFromCrdtData<T, R>([rowId])).firstOrNull;
+  }
+
+  /// Gets a list of data entries from the CRDT data table.
+  Future<List<R>> getFromCrdtData<T extends Table, R extends DataClass>(
+    List<String> rowIds,
+  ) async {
+    final tableInfo = synchronizedTables.firstWhere(
+      (t) => t is T,
+      orElse: () => throw ArgumentError('Table $T not found in synchronized tables.'),
+    );
+
+    final crdtDataEntries = await managers.crdtDataTable
+        .filter(
+          (o) => o.tblName.equals(tableInfo.actualTableName) & o.rowId.isIn(rowIds),
+        )
+        .get();
+
+    final foundRowIds = crdtDataEntries.map((e) => e.rowId).toSet();
+    return [
+      for (final rowId in foundRowIds)
+        await tableInfo.map({
+          for (final c in tableInfo.$columns)
+            c.$name: crdtDataEntries
+                .firstWhere((e) => e.rowId == rowId && e.columnName == c.$name)
+                .rawValue
+                ?.rawSqlValue,
+        }) as R,
+    ];
+  }
+}

lib/src/migrator.dart

@@ -1,6 +1,7 @@
 import 'package:drift/drift.dart';
 
 import '/src/database.dart';
+import 'crdt.dart';
 import 'triggers.dart';
 
 /// Runs migrations declared by a [MigrationStrategy].
@@ -38,7 +39,13 @@ class OfflineSyncMigrator extends Migrator {
   final List<TableInfo> synchronizedTables;
 
   /// The database instance to apply the CRDT schema to.
-  late final crdtDb = CrdtDatabase(database.executor);
+  late final crdtDb = CrdtDatabase(
+    database.executor,
+    synchronizedTables: synchronizedTables,
+  );
+
+  /// The CRDT execution instance.
+  late final crdt = OfflineSyncCrdt(crdtDb, userId: userId, nodeId: nodeId);
 
   /// Migrations for tables that are part of the CRDT system.
   final pendingMigrations = <String, CRDTTableMigration>{};

lib/src/tables/merge.dart

@@ -0,0 +1,33 @@
+import 'package:drift/drift.dart';
+
+import '../hlc/converter.dart';
+
+// Required for the generated code to know the HLC and converters.
+export '../hlc/converter.dart';
+
+/// CRDT merge HLC table.
+///
+/// This table stores the last HLC timestamp received from each node when merging
+/// changes, so that the next merges can be performed incrementally.
+///
+@TableIndex(
+  name: 'idx_crdt_merge_hlc_timestamp',
+  columns: {#userId, #hlcTimestamp},
+)
+@DataClassName('CrdtMergeHlcEntry')
+class CrdtMergeHlcTable extends Table {
+  @override
+  String get tableName => '__crdt_merge_hlc';
+
+  /// Identifier for the user or client.
+  late final userId = text()();
+
+  /// Node identifier for the client.
+  late final nodeId = text()();
+
+  /// The last HLC timestamp received from this node when merging changes.
+  late final lastReceivedHlc = text().map(hlcConverter)();
+
+  @override
+  Set<Column> get primaryKey => {userId, nodeId};
+}

pubspec.yaml

@@ -10,10 +10,13 @@ environment:
 dependencies:
   crdt: ^5.1.3
   drift: ^2.28.0
+  meta: ^1.12.0
   uuid: ^4.5.1
 
 dev_dependencies:
   build_runner: ^2.6.0
   drift_dev: ^2.28.0
   lints: ^6.0.0
+  sqlite3: any
   test: ^1.26.2
+  test_descriptor: ^2.0.2

test/crdt/crdt_canonical_time_test.dart

@@ -0,0 +1,72 @@
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:drift_offline_first/drift_offline_first.dart';
+import 'package:drift_offline_first/src/hlc/stateful.dart';
+import 'package:test/test.dart';
+
+import '../utils/crdt_context.dart';
+import '../utils/database.dart';
+
+void main() {
+  group('Given CRDT for an empty database', () {
+    late OfflineSyncCrdt crdt;
+    late String nodeId;
+
+    setUp(() async {
+      (crdt, _, nodeId) = database.crdtContext;
+    });
+
+    test(
+        'when getting the canonical time '
+        'then it returns Hlc.zero for the nodeId.', () async {
+      expect(crdt.canonicalTime, equals(Hlc.zero(nodeId)));
+    });
+  });
+
+  group('Given CRDT for a database with data', () {
+    late OfflineSyncCrdt crdt;
+    late CrdtDatabase crdtDb;
+    late Hlc createdHlc;
+
+    setUp(() async {
+      await database.managers.todosTable.create(
+        (t) => t(content: 'test', targetDate: Value(DateTime.now())),
+      );
+
+      (crdt, crdtDb, _) = database.crdtContext;
+
+      final allCrdtDataEntries = await crdtDb.managers.crdtDataTable.get();
+      createdHlc = allCrdtDataEntries.first.hlcTimestamp;
+    });
+
+    test(
+        'when getting the canonical time '
+        'then it returns the HLC timestamp of the created data.', () async {
+      expect(crdt.canonicalTime, equals(createdHlc));
+    });
+  });
+
+  group('Given CRDT for a database with data and a new data entry', () {
+    late OfflineSyncCrdt crdt;
+
+    setUp(() async {
+      await database.managers.todosTable.create(
+        (t) => t(content: 'test', targetDate: Value(DateTime.now())),
+      );
+
+      (crdt, _, _) = database.crdtContext;
+
+      await database.managers.todosTable.create(
+        (t) => t(content: 'test2', targetDate: Value(DateTime.now())),
+      );
+    });
+
+    test(
+        'when getting the canonical time '
+        'then the canonical time matches the cached HLC for the nodeId.', () async {
+      expect(
+        crdt.canonicalTime,
+        equals(StatefulHlc.cached(crdt.userId, crdt.nodeId).lastHlc),
+      );
+    });
+  });
+}

test/crdt/crdt_database_extract_test.dart

@@ -0,0 +1,76 @@
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:drift_offline_first/drift_offline_first.dart';
+import 'package:test/test.dart';
+
+import '../utils/crdt_context.dart';
+import '../utils/database.dart';
+import '../utils/tables.dart';
+
+void main() {
+  group('Given CRDT database for a database with data with every column type', () {
+    late CrdtDatabase crdtDb;
+    late String rowId;
+    late TableWithEveryColumnTypeData createdData;
+
+    setUp(() async {
+      final createdRowId = await database.managers.tableWithEveryColumnType.create(
+        (t) => t(
+          aBool: const Value(true),
+          aDateTime: Value(DateTime.now()),
+          aText: const Value('test'),
+          anInt: const Value(1),
+          anInt64: Value(BigInt.one),
+          aReal: const Value(1.0),
+          aBlob: Value(Uint8List.fromList([1, 2, 3])),
+          anIntEnum: const Value(TodoStatus.open),
+          aTextWithConverter: Value(MyCustomObject('test')),
+          aUuid: Value(UuidValue.fromString('123e4567-e89b-12d3-a456-426614174000')),
+        ),
+      );
+
+      (_, crdtDb, _) = database.crdtContext;
+      rowId = createdRowId.toString();
+      createdData = await database.managers.tableWithEveryColumnType.getSingle();
+    });
+
+    test(
+        '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<TableWithEveryColumnType,
+          TableWithEveryColumnTypeData>(rowId);
+
+      expect(data, equals(createdData));
+    });
+
+    test(
+        '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<TableWithEveryColumnType,
+          TableWithEveryColumnTypeData>('non-existent-row-id');
+
+      expect(data, isNull);
+    });
+
+    test(
+        '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<TableWithEveryColumnType, TableWithEveryColumnTypeData>(
+        [rowId, 'non-existent-row-id'],
+      );
+
+      expect(data, equals([createdData]));
+    });
+
+    test(
+        '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<TableWithoutPK, TableWithEveryColumnTypeData>(
+          [rowId],
+        ),
+        throwsA(isA<ArgumentError>()),
+      );
+    });
+  });
+}

test/crdt/crdt_get_changeset_test.dart

@@ -0,0 +1,142 @@
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:drift_offline_first/drift_offline_first.dart';
+import 'package:test/test.dart';
+
+import '../utils/crdt_context.dart';
+import '../utils/database.dart';
+
+void main() {
+  group('Given CRDT for an empty database', () {
+    late OfflineSyncCrdt crdt;
+
+    setUp(() async {
+      (crdt, _, _) = database.crdtContext;
+    });
+
+    test(
+        'when getting a changeset '
+        'then it returns an empty changeset.', () async {
+      final changeset = await crdt.getChangeset();
+
+      expect(changeset, isEmpty);
+    });
+
+    test(
+        'when adding data to the database '
+        'then the changeset contains the data.', () async {
+      await database.managers.todosTable.create(
+        (t) => t(content: 'test'),
+      );
+
+      final changeset = await crdt.getChangeset();
+      expect(changeset, isNotEmpty);
+    });
+  });
+
+  group('Given CRDT for a database with data', () {
+    late OfflineSyncCrdt crdt;
+    late CrdtDatabase crdtDb;
+    late String nodeId;
+    late String rowId;
+    late Hlc createdHlc;
+    late TodoEntry createdData;
+
+    setUp(() async {
+      final createdRowId = await database.managers.todosTable.create(
+        (t) => t(content: 'test', targetDate: Value(DateTime.now())),
+      );
+
+      (crdt, crdtDb, nodeId) = database.crdtContext;
+
+      final allCrdtDataEntries = await crdtDb.managers.crdtDataTable.get();
+      createdHlc = allCrdtDataEntries.first.hlcTimestamp;
+      rowId = createdRowId.toString();
+      createdData = await database.managers.todosTable.getSingle();
+    });
+
+    group('when getting a changeset', () {
+      group('without any filters', () {
+        test(
+            'then it returns a changeset with one entry per column of the created data.',
+            () async {
+          final changeset = await crdt.getChangeset();
+          final changesetColumns = changeset.map((e) => e.columnName).toSet();
+          final expectedColumns = database.todosTable.$columns.map((c) => c.$name);
+
+          expect(changeset.length, expectedColumns.length);
+          expect(changesetColumns, equals(expectedColumns));
+        });
+
+        test('then the raw values are set correctly.', () async {
+          final changeset = await crdt.getChangeset();
+          final rebuiltData = database.todosTable.map({
+            for (final c in database.todosTable.$columns)
+              c.$name: changeset
+                  .firstWhere((e) => e.rowId == rowId && e.columnName == c.$name)
+                  .rawValue
+                  ?.rawSqlValue,
+          });
+
+          expect(rebuiltData, equals(createdData));
+        });
+      });
+
+      test(
+          'with onlyNodeId set to the nodeId '
+          'then it returns a changeset with the data for the nodeId.', () async {
+        final changeset = await crdt.getChangeset(onlyNodeId: nodeId);
+        final nodeIds = changeset.map((e) => e.hlcTimestamp.nodeId).toSet();
+
+        expect(changeset, isNotEmpty);
+        expect(nodeIds, equals({nodeId}));
+      });
+
+      test(
+          'with onlyNodeId set to a different nodeId '
+          'then it returns an empty changeset.', () async {
+        final changeset = await crdt.getChangeset(onlyNodeId: 'wrong-node');
+
+        expect(changeset, isEmpty);
+      });
+
+      test(
+          'with exceptNodeId set to the nodeId '
+          'then it returns an empty changeset.', () async {
+        final changeset = await crdt.getChangeset(exceptNodeId: nodeId);
+
+        expect(changeset, isEmpty);
+      });
+
+      test(
+          'with exceptNodeId set to a different nodeId '
+          'then it returns a changeset with data from other nodes.', () async {
+        const otherNodeId = 'other-node';
+        final changeset = await crdt.getChangeset(exceptNodeId: otherNodeId);
+        final nodeIds = changeset.map((e) => e.hlcTimestamp.nodeId).toSet();
+
+        expect(changeset, isNotEmpty);
+        expect(nodeIds, isNot(contains(otherNodeId)));
+      });
+
+      test(
+          'with modifiedAfter set to a timestamp before the created data '
+          'then it returns a changeset with the entries from the created data.',
+          () async {
+        final earlierHlc = Hlc.zero(nodeId);
+        final changeset = await crdt.getChangeset(modifiedAfter: earlierHlc);
+
+        expect(changeset, isNotEmpty);
+        expect(changeset.map((e) => e.rowId).toSet(), equals({rowId}));
+      });
+
+      test(
+          'with modifiedAfter set to a timestamp after the created data '
+          'then it returns an empty changeset.', () async {
+        final futureHlc = createdHlc.increment();
+        final changeset = await crdt.getChangeset(modifiedAfter: futureHlc);
+
+        expect(changeset, isEmpty);
+      });
+    });
+  });
+}

test/crdt/crdt_last_modified_timestamp_test.dart

@@ -0,0 +1,78 @@
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:drift_offline_first/drift_offline_first.dart';
+import 'package:test/test.dart';
+
+import '../utils/crdt_context.dart';
+import '../utils/database.dart';
+
+void main() {
+  group('Given CRDT for an empty database', () {
+    late OfflineSyncCrdt crdt;
+
+    setUp(() async {
+      (crdt, _, _) = database.crdtContext;
+    });
+
+    test(
+        'when getting the last modified timestamp '
+        'then it returns null since the database is empty.', () async {
+      final lastModified = await crdt.getLastModified();
+      expect(lastModified, isNull);
+    });
+  });
+
+  group('Given CRDT for a database with data', () {
+    late OfflineSyncCrdt crdt;
+    late CrdtDatabase crdtDb;
+    late String nodeId;
+    late Hlc createdHlc;
+
+    setUp(() async {
+      await database.managers.todosTable.create(
+        (t) => t(content: 'test', targetDate: Value(DateTime.now())),
+      );
+
+      (crdt, crdtDb, nodeId) = database.crdtContext;
+
+      final allCrdtDataEntries = await crdtDb.managers.crdtDataTable.get();
+      createdHlc = allCrdtDataEntries.first.hlcTimestamp;
+    });
+
+    group('when getting the last modified timestamp', () {
+      test(
+          'without any filters '
+          'then it returns the HLC timestamp of the created data.', () async {
+        final lastModified = await crdt.getLastModified();
+        expect(lastModified, equals(createdHlc));
+      });
+
+      test(
+          'with onlyNodeId set to the nodeId '
+          'then it returns the HLC timestamp of the created data.', () async {
+        final lastModified = await crdt.getLastModified(onlyNodeId: nodeId);
+        expect(lastModified, equals(createdHlc));
+      });
+
+      test(
+          'with onlyNodeId set to a different nodeId '
+          'then it returns null.', () async {
+        final lastModified = await crdt.getLastModified(onlyNodeId: 'other-node');
+        expect(lastModified, isNull);
+      });
+
+      test(
+          'with exceptNodeId set to the nodeId '
+          'then it returns null.', () async {
+        final lastModified = await crdt.getLastModified(exceptNodeId: nodeId);
+        expect(lastModified, isNull);
+      });
+
+      test(
+          'with exceptNodeId set to a different nodeId '
+          'then it returns the HLC timestamp of the created data.', () async {
+        final lastModified = await crdt.getLastModified(exceptNodeId: 'other-node');
+        expect(lastModified, equals(createdHlc));
+      });
+    });
+  });
+}

test/crdt/crdt_merge_changeset_test.dart

@@ -0,0 +1,126 @@
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:drift_offline_first/drift_offline_first.dart';
+import 'package:test/test.dart';
+
+import '../utils/crdt_context.dart';
+import '../utils/database.dart';
+import '../utils/tables.dart';
+
+void main() {
+  late OfflineSyncCrdt crdt;
+  late CrdtDatabase crdtDb;
+  late String nodeId;
+
+  group('Given CRDT for an empty database and a pending changeset', () {
+    late Iterable<CrdtDataEntry> changeset;
+
+    setUp(() async {
+      (crdt, crdtDb, nodeId) = database.crdtContext;
+
+      final todo = TodoEntry(
+        id: const RowId(1),
+        content: 'test',
+        targetDate: DateTime.now(),
+      );
+
+      changeset = todo.toCrdtDataEntry(nodeId);
+    });
+
+    test(
+        'when merging a changeset '
+        'then the changeset is saved to the database.', () async {
+      await crdt.merge(changeset);
+
+      expect(await crdtDb.managers.crdtDataTable.get(), changeset);
+    });
+  });
+
+  group(
+      'Given CRDT for a database with data and a pending changeset with conflicting data and older HLC',
+      () {
+    late Iterable<CrdtDataEntry> firstChangeset;
+    late Iterable<CrdtDataEntry> lateOldChangeset;
+
+    setUp(() async {
+      (crdt, crdtDb, nodeId) = database.crdtContext;
+
+      final todo = TodoEntry(
+        id: const RowId(1),
+        content: 'test',
+        targetDate: DateTime.now(),
+      );
+
+      firstChangeset = todo.toCrdtDataEntry(nodeId);
+      await crdt.merge(firstChangeset);
+
+      lateOldChangeset = todo.toCrdtDataEntry(nodeId, hlc: Hlc.zero(nodeId));
+    });
+
+    test(
+        'when merging a changeset with older HLC '
+        'then no changes are saved to the database.', () async {
+      await crdt.merge(lateOldChangeset);
+
+      expect(await crdtDb.managers.crdtDataTable.get(), firstChangeset);
+    });
+  });
+
+  group(
+      'Given CRDT for a database with data and a pending changeset with conflicting data and some entries with newer HLC',
+      () {
+    late Iterable<CrdtDataEntry> firstChangeset;
+    late Iterable<CrdtDataEntry> newerChangeset;
+
+    setUp(() async {
+      (crdt, crdtDb, nodeId) = database.crdtContext;
+
+      final todo = TodoEntry(
+        id: const RowId(1),
+        content: 'test',
+        targetDate: DateTime.now(),
+      );
+
+      firstChangeset = todo.toCrdtDataEntry(nodeId);
+      await crdt.merge(firstChangeset);
+
+      newerChangeset =
+          todo.toCrdtDataEntry(nodeId).where((e) => e.columnName == 'content');
+    });
+
+    test(
+        'when merging a changeset with some entries with newer HLC '
+        'then the entries with newer HLC are saved to the database.', () async {
+      await crdt.merge(newerChangeset);
+
+      final expectedResultingChangeset = [
+        for (final e in firstChangeset)
+          if (e.columnName == 'content') newerChangeset.first else e,
+      ];
+
+      expect(await crdtDb.managers.crdtDataTable.get(), expectedResultingChangeset);
+    });
+  });
+}
+
+extension on TodoEntry {
+  Iterable<CrdtDataEntry> toCrdtDataEntry(String nodeId, {Hlc? hlc}) {
+    final hlcTimestamp = hlc ?? Hlc.now(nodeId);
+
+    return database.todosTable.$columns.map(
+      (c) => CrdtDataEntry(
+        userId: 'test',
+        tblName: 'todos',
+        columnName: c.$name,
+        rowId: id.toString(),
+        rawValue: (toJson()[c.$name] as Object?)?.toDriftAny(),
+        hlcTimestamp: hlcTimestamp,
+      ),
+    );
+  }
+}
+
+extension on Object {
+  DriftAny? toDriftAny() {
+    return DriftAny(this);
+  }
+}

test/crdt/crdt_save_changeset_test.dart

@@ -0,0 +1,164 @@
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:drift_offline_first/drift_offline_first.dart';
+import 'package:test/test.dart';
+import 'package:uuid/uuid.dart';
+
+import '../utils/crdt_context.dart';
+import '../utils/database.dart';
+import '../utils/tables.dart';
+
+void main() {
+  late OfflineSyncCrdt crdt;
+  late CrdtDatabase crdtDb;
+  late String otherNodeId;
+
+  group('Given CRDT for an empty database and a pending changeset', () {
+    late Iterable<CrdtDataEntry> changeset;
+    late Hlc lastHlc;
+
+    setUp(() async {
+      (crdt, crdtDb, _) = database.crdtContext;
+      otherNodeId = const Uuid().v4();
+
+      final todo = TodoEntry(
+        id: const RowId(1),
+        content: 'test',
+        targetDate: DateTime.now(),
+      );
+
+      changeset = todo.toCrdtDataEntry(otherNodeId);
+      lastHlc = Hlc.now(otherNodeId);
+    });
+
+    group('when saving a changeset', () {
+      setUp(() async {
+        await crdt.saveChangeset(changeset, [lastHlc]);
+      });
+
+      test('then the changeset is saved to the database.', () async {
+        final savedChangeset = await crdtDb.managers.crdtDataTable.get();
+        expect(savedChangeset, changeset);
+      });
+
+      test('then the merged HLCs are saved to the database.', () async {
+        final savedMergedHlcs = await crdtDb.managers.crdtMergeHlcTable.get();
+        expect(savedMergedHlcs.map((e) => e.lastReceivedHlc).single, lastHlc);
+      });
+    });
+  });
+
+  group(
+      'Given CRDT for a database with data and a pending changeset with conflicting data and older HLC',
+      () {
+    late Iterable<CrdtDataEntry> firstChangeset;
+    late Iterable<CrdtDataEntry> lateOldChangeset;
+    late Hlc lastHlc;
+    late Hlc lateOldHlc;
+
+    setUp(() async {
+      (crdt, crdtDb, _) = database.crdtContext;
+      otherNodeId = const Uuid().v4();
+
+      final todo = TodoEntry(
+        id: const RowId(1),
+        content: 'test',
+        targetDate: DateTime.now(),
+      );
+
+      lastHlc = Hlc.now(otherNodeId);
+      firstChangeset = todo.toCrdtDataEntry(otherNodeId, lastHlc);
+      await crdt.saveChangeset(firstChangeset, [lastHlc]);
+
+      lateOldHlc = Hlc.zero(otherNodeId);
+      lateOldChangeset = todo.toCrdtDataEntry(otherNodeId, lateOldHlc);
+    });
+
+    group('when saving a changeset with older HLC', () {
+      setUp(() async {
+        await crdt.saveChangeset(lateOldChangeset, [lateOldHlc]);
+      });
+
+      test('then no changes are saved to the database.', () async {
+        final savedChangeset = await crdtDb.managers.crdtDataTable.get();
+        expect(savedChangeset, firstChangeset);
+      });
+
+      test('then the last received HLC is not updated in the database.', () async {
+        final savedMergedHlcs = await crdtDb.managers.crdtMergeHlcTable.get();
+        expect(savedMergedHlcs.map((e) => e.lastReceivedHlc).single, lastHlc);
+      });
+    });
+  });
+
+  group(
+      'Given CRDT for a database with data and a pending changeset with conflicting data and some entries with newer HLC',
+      () {
+    late Iterable<CrdtDataEntry> firstChangeset;
+    late Iterable<CrdtDataEntry> newerChangeset;
+    late Hlc lastHlc;
+    late Hlc newerHlc;
+
+    setUp(() async {
+      (crdt, crdtDb, _) = database.crdtContext;
+      otherNodeId = const Uuid().v4();
+
+      final todo = TodoEntry(
+        id: const RowId(1),
+        content: 'test',
+        targetDate: DateTime.now(),
+      );
+
+      lastHlc = Hlc.now(otherNodeId);
+      firstChangeset = todo.toCrdtDataEntry(otherNodeId);
+      await crdt.saveChangeset(firstChangeset, [lastHlc]);
+
+      newerHlc = Hlc.now(otherNodeId);
+      newerChangeset = todo
+          .toCrdtDataEntry(otherNodeId, newerHlc)
+          .where((e) => e.columnName == 'content');
+    });
+
+    group('when saving a changeset with some entries with newer HLC', () {
+      setUp(() async {
+        await crdt.saveChangeset(newerChangeset, [newerHlc]);
+      });
+
+      test('then the entries with newer HLC are saved to the database.', () async {
+        final expectedResultingChangeset = [
+          for (final e in firstChangeset)
+            if (e.columnName == 'content') newerChangeset.first else e,
+        ];
+
+        expect(await crdtDb.managers.crdtDataTable.get(), expectedResultingChangeset);
+      });
+
+      test('then the last received HLC is updated in the database.', () async {
+        final savedMergedHlcs = await crdtDb.managers.crdtMergeHlcTable.get();
+        expect(savedMergedHlcs.map((e) => e.lastReceivedHlc).single, newerHlc);
+      });
+    });
+  });
+}
+
+extension on TodoEntry {
+  Iterable<CrdtDataEntry> toCrdtDataEntry(String nodeId, [Hlc? hlc]) {
+    final hlcTimestamp = hlc ?? Hlc.now(nodeId);
+
+    return database.todosTable.$columns.map(
+      (c) => CrdtDataEntry(
+        userId: 'test',
+        tblName: 'todos',
+        columnName: c.$name,
+        rowId: id.toString(),
+        rawValue: (toJson()[c.$name] as Object?)?.toDriftAny(),
+        hlcTimestamp: hlcTimestamp,
+      ),
+    );
+  }
+}
+
+extension on Object {
+  DriftAny? toDriftAny() {
+    return DriftAny(this);
+  }
+}