refactor: Improve code organization and reutilization

lib/drift_offline_first.dart

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

lib/src/crdt/extensions/entry.dart

@@ -0,0 +1,54 @@
+import 'package:drift/drift.dart';
+
+import '../../database.dart';
+import '../../hlc/converter.dart';
+
+/// Extensions for [Iterable<CrdtDataEntry>] to get the affected tables and row IDs.
+extension CrdtDataEntryExtensions on Iterable<CrdtDataEntry> {
+  /// Gets the affected tables.
+  Iterable<String> get affectedTables => map((e) => e.tblName).toSet();
+
+  /// Gets the affected row IDs.
+  Iterable<String> get affectedRowIds => map((e) => e.rowId).toSet();
+
+  /// 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);
+}

lib/src/crdt/extensions/extract.dart

@@ -18,13 +18,7 @@ extension CrdtDatabaseExtractExtensions on CrdtDatabase {
     required String tableName,
     required Iterable<String> rowIds,
   }) async {
-    final tableInfo = synchronizedTables.firstWhere(
+    final tableInfo = synchronizedTables.find(tableName);
-      (t) => t.actualTableName == tableName,
-      orElse: () => throw ArgumentError(
-        'Table $tableName not found in synchronized tables.',
-      ),
-    );
-
     final crdtDataEntries = await managers.crdtDataTable
         .filter((o) => o.tblName.equals(tableName) & o.rowId.isIn(rowIds))
         .get();

lib/src/database.dart

@@ -4,6 +4,7 @@ import 'tables/compensation.dart';
 import 'tables/control.dart';
 import 'tables/data.dart';
 import 'tables/merge.dart';
+import 'utils/sql_builder.dart';
 
 part 'database.g.dart';
 
@@ -18,11 +19,25 @@ part 'database.g.dart';
 )
 class CrdtDatabase extends _$CrdtDatabase {
   /// Creates a new instance of [CrdtDatabase] with the provided [executor].
-  CrdtDatabase(super.e, {required this.synchronizedTables});
+  CrdtDatabase(
+    super.e, {
+    required this.userId,
+    required this.nodeId,
+    required this.synchronizedTables,
+  });
+
+  /// The user ID for the CRDT system.
+  final String userId;
+
+  /// The node ID for the CRDT system.
+  final String nodeId;
 
   /// Tables to be synchronized with CRDT.
   final List<TableInfo> synchronizedTables;
 
+  /// The SQL builder for the CRDT data table.
+  late final sqlBuilder = CrdtDataSqlBuilder(this);
+
   /// Whether the database is using the SQLite or Postgres dialect.
   bool get isPostgres => executor.dialect == SqlDialect.postgres;
 
@@ -31,4 +46,21 @@ 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> {
+  /// Gets the table names of the synchronized tables.
+  Iterable<String> get tableNames => map((t) => t.actualTableName);
+
+  /// Gets the synchronized table information for the given table name.
+  ///
+  /// If the table is not found, an [ArgumentError] is thrown.
+  TableInfo find(String tableName) {
+    return firstWhere(
+      (t) => t.actualTableName == tableName,
+      orElse: () => throw ArgumentError(
+        'Table "$tableName" not found in synchronized tables.',
+      ),
+    );
+  }
 }

lib/src/migrator.dart

@@ -41,6 +41,8 @@ class OfflineSyncMigrator extends Migrator {
   /// The database instance to apply the CRDT schema to.
   late final crdtDb = CrdtDatabase(
     database.executor,
+    userId: userId,
+    nodeId: nodeId,
     synchronizedTables: synchronizedTables,
   );
 
@@ -48,7 +50,7 @@ class OfflineSyncMigrator extends Migrator {
   late final crdt = OfflineSyncCrdt(crdtDb, userId: userId, nodeId: nodeId);
 
   /// Migrations for tables that are part of the CRDT system.
-  final pendingMigrations = <String, CRDTTableMigration>{};
+  final pendingMigrations = <String, CrdtTableMigration>{};
 
   /// Eagerly validates that all synchronized tables are supported.
   ///
@@ -69,10 +71,10 @@ class OfflineSyncMigrator extends Migrator {
   ///
   /// If the table is not in the list of synchronized tables, a new migration
   /// will be created.
-  CRDTTableMigration _register(String tableName) {
+  CrdtTableMigration _register(String tableName) {
     return pendingMigrations.putIfAbsent(
       tableName,
-      () => CRDTTableMigration(tableName),
+      () => CrdtTableMigration(tableName),
     );
   }
 
@@ -89,6 +91,9 @@ class OfflineSyncMigrator extends Migrator {
         // TODO: Fix this partitioning, since RANGE can not be used for discrete values.
         'PARTITION BY RANGE (${crdtTable.userId.name}, ${crdtTable.tblName.name});',
       );
+
+      // TODO: Alter all ON DELETE FK constraints from RESTRICT to NO ACTION.
+      // TODO: Alter all FK constraints to be deferrable initially deferred.
     }
 
     for (final table in synchronizedTables) {
@@ -98,7 +103,8 @@ class OfflineSyncMigrator extends Migrator {
 
   Future<void> _createTriggers(TableInfo table) async {
     final triggerCreator = switch (database.executor.dialect) {
-      SqlDialect.sqlite => Sqlite3OfflineSyncTriggers(this),
+      SqlDialect.sqlite => Sqlite3OfflineSyncTriggers(crdtDb),
+      // TODO: Implement the triggers for the Postgres dialect.
       _ => throw UnsupportedError('Unsupported dialect: ${database.executor.dialect}'),
     };
     final triggers = triggerCreator.generateCreateTriggerStatements(table);
@@ -131,7 +137,7 @@ class OfflineSyncMigrator extends Migrator {
     final table = migration.affectedTable;
     pendingMigrations.putIfAbsent(
       table.actualTableName,
-      () => CRDTTableMigration.from(migration),
+      () => CrdtTableMigration.from(migration),
     );
 
     final existingTriggers = await database.getCrdtTriggers();
@@ -187,9 +193,9 @@ class OfflineSyncMigrator extends Migrator {
 }
 
 /// A migration for a table that is part of the CRDT system.
-class CRDTTableMigration {
+class CrdtTableMigration {
-  /// Creates a new instance of [CRDTTableMigration].
+  /// Creates a new instance of [CrdtTableMigration].
-  CRDTTableMigration(this.tableName);
+  CrdtTableMigration(this.tableName);
 
   /// The table that is being migrated.
   final String tableName;
@@ -204,7 +210,7 @@ class CRDTTableMigration {
   String? oldName;
 
   /// The columns that were added.
-  final columns = <String, CRDTColumnMigration>{};
+  final columns = <String, CrdtColumnMigration>{};
 
   /// Whether the table was modified.
   bool get modified => oldName != null || columns.isNotEmpty;
@@ -213,14 +219,14 @@ class CRDTTableMigration {
   ///
   /// If the column is not in the list of synchronized columns, a new migration
   /// will be created.
-  CRDTColumnMigration onColumn(String columnName) {
+  CrdtColumnMigration onColumn(String columnName) {
-    return columns.putIfAbsent(columnName, () => CRDTColumnMigration(columnName));
+    return columns.putIfAbsent(columnName, () => CrdtColumnMigration(columnName));
   }
 
-  /// Creates a new instance of [CRDTTableMigration] from a [TableMigration].
+  /// Creates a new instance of [CrdtTableMigration] from a [TableMigration].
-  static CRDTTableMigration from(TableMigration migration) {
+  static CrdtTableMigration from(TableMigration migration) {
     final table = migration.affectedTable;
-    final tableMigration = CRDTTableMigration(table.actualTableName);
+    final tableMigration = CrdtTableMigration(table.actualTableName);
     for (final column in migration.newColumns) {
       // TODO: Handle the column transformer.
       tableMigration.onColumn(column.name);
@@ -230,9 +236,9 @@ class CRDTTableMigration {
 }
 
 /// A migration for a table that is part of the CRDT system.
-class CRDTColumnMigration {
+class CrdtColumnMigration {
-  /// Creates a new instance of [CRDTColumnMigration].
+  /// Creates a new instance of [CrdtColumnMigration].
-  CRDTColumnMigration(this.columnName);
+  CrdtColumnMigration(this.columnName);
 
   /// The column that is being migrated.
   final String columnName;

lib/src/triggers.dart

@@ -1,6 +1,6 @@
 import 'package:drift/drift.dart';
 
-import '/src/migrator.dart';
+import 'utils/sql_builder.dart';
 
 /// The name of the HLC function used in the database.
 const nextHlcFunction = 'next_hlc_timestamp';
@@ -13,25 +13,9 @@ enum Operation { insert, update, delete }
 // now are no longer universal as the one developed originally.
 
 /// Provides the CRDT control trigger statements for the database.
-abstract class OfflineSyncTriggers {
+abstract class OfflineSyncTriggers extends CrdtDataSqlBuilder {
-  /// Creates a new instance of [OfflineSyncTriggers] with the provided [migrator].
+  /// Creates a new instance of [OfflineSyncTriggers].
-  const OfflineSyncTriggers(this.migrator);
+  OfflineSyncTriggers(super.crdtDb);
-
-  /// The migrator instance to apply the CRDT schema to.
-  final OfflineSyncMigrator migrator;
-
-  /// Prefix for all CRDT-related entities.
-  String get _prefix => '__crdt';
-
-  /// Name of the column that stores the information about whether the row is deleted.
-  String get isDeletedColumnName => '${_prefix}_is_deleted';
-
-  /// Name of the CRDT data table.
-  String get _crdtDataTableName => migrator.crdtDb.crdtDataTable.actualTableName;
-
-  /// The list of columns in the CRDT data table.
-  Iterable<String> get _crdtDataAllColumns =>
-      migrator.crdtDb.crdtDataTable.$columns.quotedNames;
 
   /// Wraps the provided SQL statement as a CREATE TRIGGER statement.
   ///
@@ -45,7 +29,7 @@ abstract class OfflineSyncTriggers {
 
   /// Gets the name of the trigger for a specific operation on a table.
   String _getTriggerName(TableInfo<Table, Object?> table, Operation operation) =>
-      '${_prefix}__${table.actualTableName}__${operation.name}';
+      '${prefix}__${table.actualTableName}__${operation.name}';
 
   /// Returns the SQL statement to insert a new row into the CRDT data table.
   ///
@@ -63,8 +47,10 @@ abstract class OfflineSyncTriggers {
   }) {
     final tableName = table.actualTableName;
 
-    final uniqueRowId =
+    final uniqueRowId = getUniqueRowId(
-        table.getUniqueRowId(tableAlias: operation == Operation.insert ? 'NEW' : 'OLD');
+      table,
+      tableAlias: operation == Operation.insert ? 'NEW' : 'OLD',
+    );
 
     final columnNamesWithoutPrimaryKey = operation == Operation.delete
         ? [isDeletedColumnName]
@@ -91,18 +77,18 @@ abstract class OfflineSyncTriggers {
         operation == Operation.update ? '\n  WHERE (value_changed = TRUE)' : '';
 
     return '''
-  INSERT OR REPLACE INTO $_crdtDataTableName (
+  INSERT OR REPLACE INTO $crdtDataTableName (
-    ${_crdtDataAllColumns.join(', ')}
+    ${crdtDataAllColumns.join(', ')}
   )
   SELECT
-    '${migrator.userId}' AS "user_id",
+    '${crdtDb.userId}' AS "user_id",
     '$tableName' AS "table_name",
     column_name,
     $uniqueRowId AS "row_id",
     "hlc_timestamp",
     raw_value
   FROM (
-    SELECT $nextHlcFunction('${migrator.userId}', '${migrator.nodeId}') AS "hlc_timestamp"
+    SELECT $nextHlcFunction('${crdtDb.userId}', '${crdtDb.nodeId}') AS "hlc_timestamp"
   ), (
 ${columnInserts.join('\n    UNION ALL\n')}
   )$whereClause;''';
@@ -128,7 +114,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.migrator);
+  Sqlite3OfflineSyncTriggers(super.crdtDb);
 
   @override
   String wrapAsCreateTriggerStatement({
@@ -145,30 +131,6 @@ $innerSql
 END;''';
 }
 
-extension on TableInfo {
-  /// Returns an expression that uniquely identifies a row in the table.
-  ///
-  /// If the table has a single primary key column, the expression will be the
-  /// column value. If it has multiple primary key columns, the values for all
-  /// will be concatenated using the `_||_` operator.
-  String getUniqueRowId({required String tableAlias}) {
-    if ($primaryKey.isEmpty) {
-      throw StateError('Table $actualTableName has no primary key and is therefore '
-          'not supported to track changes to it.');
-    }
-    if ($primaryKey.length == 1) {
-      return '$tableAlias."${$primaryKey.first.$name}"';
-    }
-    final columnNames = $primaryKey.map((c) => '$tableAlias."${c.$name}"').join(', ');
-    return "CONCAT_WS('_||_', $columnNames)";
-  }
-}
-
-extension on Iterable<GeneratedColumn> {
-  /// Returns the column names with quotes for safer SQL queries.
-  Iterable<String> get quotedNames => map((c) => '"${c.$name}"');
-}
-
 /// Extension methods for the [GeneratedDatabase] class to get the CRDT trigger names.
 extension GetDatabaseCrdtTriggers on GeneratedDatabase {
   /// Returns the names of the CRDT triggers in the database.

lib/src/utils/sql_builder.dart

@@ -0,0 +1,58 @@
+import 'package:drift/drift.dart';
+
+import '/src/database.dart';
+
+/// Provides the base building blocks for SQL statements to interact with the
+/// CRDT data table.
+class CrdtDataSqlBuilder {
+  /// Creates a new instance of [CrdtDataSqlBuilder].
+  const CrdtDataSqlBuilder(this.crdtDb);
+
+  /// The migrator instance to apply the CRDT schema to.
+  final CrdtDatabase crdtDb;
+
+  /// Prefix for all CRDT-related entities.
+  String get prefix => '__crdt';
+
+  /// Separator for joining primary key columns in the row ID.
+  String get rowIdSeparator => '||';
+
+  /// Name of the column that stores the information about whether the row is deleted.
+  String get isDeletedColumnName => '${prefix}_is_deleted';
+
+  /// Name of the CRDT data table.
+  String get crdtDataTableName => crdtDb.crdtDataTable.actualTableName;
+
+  /// The list of quoted column names in the CRDT data table.
+  Iterable<String> get crdtDataAllColumns => crdtDb.crdtDataTable.$columns.quotedNames;
+
+  /// The list of quoted primary key column names in the CRDT data table.
+  Iterable<String> get crdtDataPrimaryKeyColumns =>
+      crdtDb.crdtDataTable.$primaryKey.quotedNames;
+
+  /// Returns an expression that uniquely identifies a row in the table.
+  ///
+  /// If the table has a single primary key column, the expression will be the
+  /// column value. If it has multiple primary key columns, the values for all
+  /// will be concatenated using the separator.
+  String getUniqueRowId(TableInfo table, {required String tableAlias}) {
+    if (table.$primaryKey.isEmpty) {
+      throw StateError(
+        'Table ${table.actualTableName} has no primary key and is therefore '
+        'not supported to track changes to it.',
+      );
+    }
+    if (table.$primaryKey.length == 1) {
+      return '$tableAlias."${table.$primaryKey.first.$name}"';
+    }
+    final columnNames =
+        table.$primaryKey.map((c) => '$tableAlias."${c.$name}"').join(', ');
+    return "CONCAT_WS('$rowIdSeparator', $columnNames)";
+  }
+}
+
+/// Extension methods for the [Iterable<GeneratedColumn>] class to get the column names with quotes.
+extension GetColumnNamesWithQuotes on Iterable<GeneratedColumn> {
+  /// Returns the column names with quotes for safer SQL queries.
+  Iterable<String> get quotedNames => map((c) => '"${c.$name}"');
+}

test/crdt/crdt_data_convert_test.dart

@@ -0,0 +1,94 @@
+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 a data object with every column type', () {
+    late CrdtDatabase crdtDb;
+    late TableWithEveryColumnTypeData createdData;
+    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'),
+      );
+
+      (_, crdtDb, _) = database.crdtContext;
+      testHlc = Hlc.now(crdtDb.nodeId);
+    });
+
+    group('when converting the data to CRDT data entries', () {
+      late Iterable<CrdtDataEntry> entries;
+
+      setUp(() async {
+        entries = createdData.toCrdtDataEntry(crdtDb, testHlc).toList();
+      });
+
+      test('then it creates an entry for each column.', () async {
+        expect(
+          entries.map((e) => e.columnName).toSet(),
+          equals(
+            database.tableWithEveryColumnType.$columns.map((c) => c.$name).toSet(),
+          ),
+        );
+      });
+
+      test('then each entry has the correct user ID.', () async {
+        expect(entries.map((e) => e.userId), everyElement(equals(crdtDb.userId)));
+      });
+
+      test('then each entry has the correct table name.', () async {
+        expect(
+          entries.map((e) => e.tblName),
+          everyElement(equals('table_with_every_column_type')),
+        );
+      });
+
+      test('then each entry has the correct HLC timestamp.', () async {
+        expect(entries.map((e) => e.hlcTimestamp), everyElement(equals(testHlc)));
+      });
+
+      test('then all entries have expected column names and values.', () async {
+        final entriesMap = {
+          for (final entry in entries) entry.columnName: entry.rawValue,
+        };
+
+        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));
+        expect(
+          entriesMap['an_int_enum'],
+          equals(createdData.anIntEnum?.index.driftAny),
+        );
+        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);
+}

test/crdt/crdt_data_extract_test.dart

@@ -0,0 +1,81 @@
+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<TableWithEveryColumnTypeData>(
+        tableName: 'table_with_every_column_type',
+        rowId: 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<TableWithEveryColumnTypeData>(
+        tableName: 'table_with_every_column_type',
+        rowId: '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<TableWithEveryColumnTypeData>(
+        tableName: 'table_with_every_column_type',
+        rowIds: [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<TableWithEveryColumnTypeData>(
+          tableName: 'table_without_pk',
+          rowIds: [rowId],
+        ),
+        throwsA(isA<ArgumentError>()),
+      );
+    });
+  });
+}

test/crdt/crdt_merge_changeset_test.dart

@@ -1,4 +1,3 @@
-import 'package:drift/drift.dart' hide isNotNull, isNull;
 import 'package:drift_offline_first/drift_offline_first.dart';
 import 'package:test/test.dart';
 
@@ -23,7 +22,7 @@ void main() {
         targetDate: DateTime.now(),
       );
 
-      changeset = todo.toCrdtDataEntry(nodeId);
+      changeset = todo.toCrdtDataEntry(crdtDb, Hlc.now(nodeId));
     });
 
     test(
@@ -50,10 +49,10 @@ void main() {
         targetDate: DateTime.now(),
       );
 
-      firstChangeset = todo.toCrdtDataEntry(nodeId);
+      firstChangeset = todo.toCrdtDataEntry(crdtDb, Hlc.now(nodeId));
       await crdt.merge(firstChangeset);
 
-      lateOldChangeset = todo.toCrdtDataEntry(nodeId, hlc: Hlc.zero(nodeId));
+      lateOldChangeset = todo.toCrdtDataEntry(crdtDb, Hlc.zero(nodeId));
     });
 
     test(
@@ -80,11 +79,12 @@ void main() {
         targetDate: DateTime.now(),
       );
 
-      firstChangeset = todo.toCrdtDataEntry(nodeId);
+      firstChangeset = todo.toCrdtDataEntry(crdtDb, Hlc.now(nodeId));
       await crdt.merge(firstChangeset);
 
-      newerChangeset =
+      newerChangeset = todo
-          todo.toCrdtDataEntry(nodeId).where((e) => e.columnName == 'content');
+          .toCrdtDataEntry(crdtDb, Hlc.now(nodeId))
+          .where((e) => e.columnName == 'content');
     });
 
     test(
@@ -101,26 +101,3 @@ void main() {
     });
   });
 }
-
-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

@@ -1,4 +1,3 @@
-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';
@@ -26,7 +25,7 @@ void main() {
         targetDate: DateTime.now(),
       );
 
-      changeset = todo.toCrdtDataEntry(otherNodeId);
+      changeset = todo.toCrdtDataEntry(crdtDb, Hlc.now(otherNodeId));
       lastHlc = Hlc.now(otherNodeId);
     });
 
@@ -66,11 +65,11 @@ void main() {
       );
 
       lastHlc = Hlc.now(otherNodeId);
-      firstChangeset = todo.toCrdtDataEntry(otherNodeId, lastHlc);
+      firstChangeset = todo.toCrdtDataEntry(crdtDb, lastHlc);
       await crdt.saveChangeset(firstChangeset, [lastHlc]);
 
       lateOldHlc = Hlc.zero(otherNodeId);
-      lateOldChangeset = todo.toCrdtDataEntry(otherNodeId, lateOldHlc);
+      lateOldChangeset = todo.toCrdtDataEntry(crdtDb, lateOldHlc);
     });
 
     group('when saving a changeset with older HLC', () {
@@ -109,12 +108,12 @@ void main() {
       );
 
       lastHlc = Hlc.now(otherNodeId);
-      firstChangeset = todo.toCrdtDataEntry(otherNodeId);
+      firstChangeset = todo.toCrdtDataEntry(crdtDb, lastHlc);
       await crdt.saveChangeset(firstChangeset, [lastHlc]);
 
       newerHlc = Hlc.now(otherNodeId);
       newerChangeset = todo
-          .toCrdtDataEntry(otherNodeId, newerHlc)
+          .toCrdtDataEntry(crdtDb, newerHlc)
           .where((e) => e.columnName == 'content');
     });
 
@@ -139,26 +138,3 @@ void main() {
     });
   });
 }
-
-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/triggers/trigger_generation_test.dart

@@ -16,7 +16,7 @@ void main() {
       late List<String> triggers;
 
       setUp(() async {
-        final generator = Sqlite3OfflineSyncTriggers(migrator);
+        final generator = Sqlite3OfflineSyncTriggers(migrator.crdtDb);
         triggers = generator.generateCreateTriggerStatements(database.todosTable);
       });