refactor: Move the Hlc generation from Dart to the database (#20)

benchmark/utils/database.dart

@@ -25,10 +25,7 @@ class CrdtBenchmarkDatabase extends _$CrdtBenchmarkDatabase {
     }
 
     return CrdtBenchmarkDatabase(
-      NativeDatabase.opened(
+      NativeDatabase.opened(sqlite3.open(file.path)),
-        sqlite3.open(file.path),
-        setup: crdtEnabled ? registerHlcFunction : null,
-      ),
       crdtEnabled: crdtEnabled,
     );
   }

lib/drift_offline_sync.dart

@@ -11,7 +11,6 @@ export 'src/database/extensions/entry.dart';
 export 'src/database/extensions/extract.dart';
 export 'src/database/extensions/sync.dart';
 export 'src/database/extensions/triggers.dart';
-export 'src/database/setup.dart';
 export 'src/hlc/converter.dart';
 export 'src/hlc/exceptions.dart';
 export 'src/hlc/hlc.dart';

lib/src/crdt/base.dart

@@ -4,7 +4,6 @@ import 'package:meta/meta.dart';
 
 import '../database/database.dart';
 import '../hlc/hlc.dart';
-import '../hlc/stateful.dart';
 
 /// Base class for CRDT implementations.
 ///
@@ -21,11 +20,11 @@ abstract class CrdtBase {
   /// The node ID of the CRDT.
   final String nodeId;
 
-  /// The stateful canonical time of the CRDT.
+  /// The current canonical time from the database (in-SQLite HLC state).
-  late final StatefulHlc _statefulCanonicalTime = StatefulHlc.cached(userId, nodeId);
+  Future<Hlc> getCanonicalTime();
 
-  /// The canonical time of the CRDT.
+  /// Merge the [maxMergedHlc] into the canonical time.
-  Hlc get canonicalTime => _statefulCanonicalTime.lastHlc;
+  Future<void> mergeCanonicalTime(Hlc maxMergedHlc);
 
   /// Returns the last modified timestamp, optionally filtering for or against a
   /// specific node id. Useful to get "modified since" timestamps for executing
@@ -69,6 +68,10 @@ abstract class CrdtBase {
   );
 
   /// Merge [changeset] with the local dataset.
+  ///
+  /// This operation must be done within a transaction to ensure that no other
+  /// operation modifies the data between the save, invariants preservation and
+  /// the merge of remote HLCs into the canonical time.
   Future<void> merge(Iterable<CrdtDataEntry> changeset) async {
     if (changeset.isEmpty) return;
 
@@ -76,7 +79,7 @@ abstract class CrdtBase {
     await saveChangeset(changeset, receivedHlcs);
 
     if (receivedHlcs.isEmpty) return;
-    _statefulCanonicalTime.merge(receivedHlcs.max);
+    await mergeCanonicalTime(receivedHlcs.max);
   }
 
   /// Extracts the last received HLCs from the changeset.

lib/src/crdt/crdt.dart

@@ -1,7 +1,8 @@
 import 'package:drift/drift.dart';
 
 import '../database/database.dart';
-import '../hlc/hlc.dart';
+import '../database/extensions/entry.dart';
+import '../hlc/converter.dart';
 import 'base.dart';
 
 /// A CRDT implementation for offline-first synchronization.
@@ -13,6 +14,56 @@ class OfflineSyncCrdt extends CrdtBase {
 
   final CrdtDatabase _db;
 
+  @override
+  Future<Hlc> getCanonicalTime() async {
+    final row = await _db.managers.crdtHlcStateTable
+        .filter((o) => o.userId.equals(userId))
+        .getSingleOrNull();
+
+    return row?.toHlc(nodeId) ?? Hlc.zero(nodeId);
+  }
+
+  @override
+  Future<void> mergeCanonicalTime(Hlc maxMergedHlc) async {
+    await _db.managers.crdtHlcStateTable.create(
+      (t) => t(
+        userId: userId,
+        lastTimestamp: maxMergedHlc.unixTimestamp,
+        counter: maxMergedHlc.counter,
+      ),
+      mode: InsertMode.insertOrReplace,
+      onConflict: DoUpdate.withExcluded(
+        (old, excluded) => CrdtHlcStateTableCompanion.custom(
+          lastTimestamp: CaseWhenExpression(
+            // Combined with the `orElse`, this is the same as a max() function,
+            // but it works for all databases. Otherwise, we would need to check
+            // the database dialect to use either `MAX` or `GREATEST`.
+            cases: [
+              excluded.lastTimestamp
+                  .isBiggerOrEqual(old.lastTimestamp)
+                  .then(excluded.lastTimestamp),
+            ],
+            orElse: old.lastTimestamp,
+          ),
+          counter: CaseWhenExpression(
+            cases: [
+              // If one timestamp is greater, its counter should be persisted.
+              excluded.lastTimestamp
+                  .isBiggerThan(old.lastTimestamp)
+                  .then(excluded.counter),
+              // Otherwise, if the timestamps are equal, the counter should be
+              // the greater counter.
+              (excluded.lastTimestamp.equalsExp(old.lastTimestamp) &
+                      excluded.counter.isBiggerThan(old.counter))
+                  .then(excluded.counter),
+            ],
+            orElse: old.counter,
+          ),
+        ),
+      ),
+    );
+  }
+
   @override
   // TODO: Remove this method lately if it is not needed - as it seems it will.
   Future<Hlc?> getLastModified({String? onlyNodeId, String? exceptNodeId}) async {
@@ -62,7 +113,9 @@ class OfflineSyncCrdt extends CrdtBase {
         condition &= o.hlcTimestamp.column.contains(exceptNodeId).not();
       }
       if (modifiedAfter != null) {
-        condition &= o.hlcTimestamp.column.isBiggerThanValue(modifiedAfter.toString());
+        condition &= o.hlcTimestamp.column.isBiggerThanValue(
+          hlcConverter.toSql(modifiedAfter),
+        );
       }
       return condition;
     });
@@ -133,3 +186,10 @@ class OfflineSyncCrdt extends CrdtBase {
     );
   }
 }
+
+extension on Expression<bool> {
+  CaseWhen<bool, T> then<T extends Object>(Expression<T> value) => CaseWhen(
+    this,
+    then: value,
+  );
+}

lib/src/database/database.dart

@@ -3,6 +3,7 @@ import 'package:drift/drift.dart';
 import '../utils/sql_builder.dart';
 import 'tables/control.dart';
 import 'tables/data.dart';
+import 'tables/hlc.dart';
 import 'tables/merge.dart';
 
 part 'database.g.dart';
@@ -12,6 +13,7 @@ part 'database.g.dart';
   tables: [
     CrdtDataTable,
     CrdtControlTable,
+    CrdtHlcStateTable,
     CrdtMergeHlcTable,
   ],
 )

lib/src/database/extensions/convert.dart

@@ -1,6 +1,6 @@
 import 'package:drift/drift.dart';
 
-import '../../hlc/converter.dart';
+import '../../hlc/hlc.dart';
 import '../database.dart';
 
 /// Extensions for [CrdtDatabase] to convert data to [CrdtDataEntry].

lib/src/database/extensions/entry.dart

@@ -1,6 +1,16 @@
-import '../../hlc/converter.dart';
+import '../../hlc/hlc.dart';
 import '../database.dart';
 
+/// Extensions for the [CrdtHlcEntry] class.
+extension CrdtHlcEntryExtensions on CrdtHlcEntry {
+  /// Returns the HLC. If the [nodeId] is not provided, it will use the [userId].
+  Hlc toHlc([String? nodeId]) => Hlc(
+    DateTime.fromMillisecondsSinceEpoch(lastTimestamp, isUtc: true),
+    counter,
+    nodeId ?? userId,
+  );
+}
+
 /// Extensions for the [CrdtDataEntry] class.
 extension CrdtDataEntryUnwrapExtensions on CrdtDataEntry {
   /// Returns the unwrapped raw value, if any.

lib/src/database/setup.dart

@@ -1,57 +0,0 @@
-import 'package:drift/native.dart';
-
-import '/src/hlc/stateful.dart';
-import 'tables/data.dart';
-import 'triggers.dart';
-
-/// The database setup function to register the HLC function.
-///
-/// Will register the `next_hlc_timestamp` function in the SQLite3 database for usage
-/// in the triggers that keep track of changes in the database.
-///
-/// If using Dart only, must be passed to the [NativeDatabase] constructor as `setup`
-/// parameter. If using Flutter, this function must be passed to the `driftDatabase`
-/// function as `setup` parameter.
-DatabaseSetup get registerHlcFunction {
-  return (database) {
-    database.createFunction(
-      functionName: nextHlcFunction,
-      function: (args) {
-        if (args.isEmpty) throw _MissingArgumentsError();
-        if (args.length == 1) throw _MissingNodeIdError();
-        if (args.length > 2) throw _TooManyArgumentsError();
-
-        final userId = args[0] as String?;
-        if (userId == null) throw _MissingUserIdError();
-
-        final nodeId = args[1] as String?;
-        if (nodeId == null) throw _MissingNodeIdError();
-
-        return StatefulHlc.cached(userId, nodeId).increment().toString();
-      },
-      // argumentCount: const AllowedArgumentCount(2),
-      deterministic: false,
-      directOnly: false,
-    );
-
-    // Load the last HLC for the user from the control table.
-    try {
-      database.select(CrdtDataTable.getLastHlcTimestampQuery()).forEach((row) {
-        final userId = row.values.first! as String;
-        final lastHlcTimestamp = hlcConverter.fromSql(row.values.last! as String);
-        StatefulHlc.initialize(userId, lastHlcTimestamp);
-      });
-    } on Exception catch (_) {
-      // Ignore errors, as the HLC will be initialized to zero.
-      // This is expected to happen on a new database.
-    }
-  };
-}
-
-class _MissingArgumentsError extends ArgumentError {}
-
-class _MissingUserIdError extends _MissingArgumentsError {}
-
-class _MissingNodeIdError extends _MissingArgumentsError {}
-
-class _TooManyArgumentsError extends ArgumentError {}

lib/src/database/tables/hlc.dart

@@ -0,0 +1,23 @@
+import 'package:drift/drift.dart';
+
+/// HLC state table.
+///
+/// Stores per-user last timestamp and counter for Hybrid Logical Clock state.
+///
+@DataClassName('CrdtHlcEntry')
+class CrdtHlcStateTable extends Table {
+  @override
+  String get tableName => '__hlc_state';
+
+  /// Identifier for the user or client.
+  late final userId = text()();
+
+  /// Last HLC timestamp in milliseconds since epoch (logical time component).
+  late final lastTimestamp = integer()();
+
+  /// Counter for events at the same logical time (count component).
+  late final counter = integer()();
+
+  @override
+  Set<Column> get primaryKey => {userId};
+}

lib/src/database/triggers.dart

@@ -3,16 +3,10 @@ import 'package:drift/drift.dart';
 import '../migrator.dart';
 import '../utils/sql_builder.dart';
 
-/// The name of the HLC function used in the database.
-const nextHlcFunction = 'next_hlc_timestamp';
-
 /// The operation that was performed on a row.
 // ignore: public_member_api_docs
 enum Operation { insert, update, delete }
 
-// TODO: The [OfflineSyncTriggers] class should be reworked since the SQL statements
-// now are no longer universal as the one developed originally.
-
 /// Provides the CRDT control trigger statements for the database.
 abstract class OfflineSyncTriggers extends CrdtDataSqlBuilder {
   /// Creates a new instance of [OfflineSyncTriggers].
@@ -39,16 +33,61 @@ abstract class OfflineSyncTriggers extends CrdtDataSqlBuilder {
       '${prefix}__${table.actualTableName}__${operation.name}';
 
   /// Returns the SQL statement to insert a new row into the CRDT data table.
-  ///
+  String getInsertStatement({
-  /// This statement is compatible with all major SQL dialects like SQLite,
+    required TableInfo<Table, Object?> table,
-  /// Postgres, MySQL, SQL Server, etc. So the dialect-specific subclass only
+    required Operation operation,
-  /// needs to wrap it in the appropriate CREATE TRIGGER statement.
+  });
-  ///
+
-  /// For some dialects, it will only be needed to enable support for quoted
+  /// Returns the SQL statements to create the triggers for the changelog table.
-  /// identifiers that are used in the statement for maximum compatibility. On
+  List<String> generateCreateTriggerStatements(TableInfo<Table, Object?> table) {
-  /// MySQL this is done by setting the `SQL_MODE` to `ANSI_QUOTES` and on SQL
+    return [
-  /// Server this is done by setting the `QUOTED_IDENTIFIER` option to `ON`.
+      for (final op in Operation.values)
-  String _getInsertStatement({
+        wrapAsCreateTriggerStatement(
+          operation: op,
+          tableName: table.actualTableName,
+          triggerName: _getTriggerName(table, op),
+          innerSql: getInsertStatement(
+            table: table,
+            operation: op,
+          ),
+        ),
+    ];
+  }
+
+  /// Returns extra statements to execute once before the triggers are created.
+  List<String> generateSetupStatements() => const [];
+}
+
+/// Provides the CRDT control trigger statements for the SQLite3 database.
+class Sqlite3OfflineSyncTriggers extends OfflineSyncTriggers {
+  /// Creates a new instance of [Sqlite3OfflineSyncTriggers].
+  Sqlite3OfflineSyncTriggers(super.crdtDb, [super.conflictingBias]);
+
+  /// Name of the view used to advance HLC via INSERT (INSTEAD OF trigger).
+  static const hlcCanonicalViewName = '__hlc_canonical_view';
+
+  @override
+  String wrapAsCreateTriggerStatement({
+    required Operation operation,
+    required String tableName,
+    required String triggerName,
+    required String innerSql,
+  }) =>
+      '''
+CREATE TRIGGER IF NOT EXISTS "$triggerName"
+AFTER ${operation.name.toUpperCase()} ON "$tableName"
+WHEN (
+  SELECT MAX("crdt_triggers_on")
+  FROM "__crdt_control"
+  WHERE ("user_id" = '${crdtDb.userId}')
+    AND ("node_id" = '${crdtDb.nodeId}')
+) = TRUE
+BEGIN
+$innerSql
+END;''';
+
+  @override
+  String getInsertStatement({
     required TableInfo<Table, Object?> table,
     required Operation operation,
   }) {
@@ -85,6 +124,9 @@ abstract class OfflineSyncTriggers extends CrdtDataSqlBuilder {
         : 'TRUE';
 
     return '''
+  INSERT INTO "$hlcCanonicalViewName" ("user_id")
+  VALUES ('${crdtDb.userId}');
+
   INSERT INTO "$crdtDataTableName" (
     ${crdtDataAllColumns.join(', ')}
   )
@@ -93,11 +135,13 @@ abstract class OfflineSyncTriggers extends CrdtDataSqlBuilder {
     '$tableName' AS "table_name",
     "column_name",
     $uniqueRowId AS "row_id",
-    "hlc_timestamp",
+    (
+      SELECT (${_getHlcJsonExprSql()})
+      FROM "$hlcStateTableName"
+      WHERE "user_id" = '${crdtDb.userId}'
+    ) AS "hlc_timestamp",
     raw_value
   FROM (
-    SELECT $nextHlcFunction('${crdtDb.userId}', '${crdtDb.nodeId}') AS "hlc_timestamp"
-  ), (
 ${columnInserts.join('\n    UNION ALL\n')}
   )
   WHERE $whereClause
@@ -107,49 +151,57 @@ ${columnInserts.join('\n    UNION ALL\n')}
     "raw_value" = EXCLUDED."raw_value";''';
   }
 
-  /// Returns the SQL statements to create the triggers for the changelog table.
+  /// Returns SQL fragment that formats (last_timestamp, counter, node_id) in the
-  List<String> generateCreateTriggerStatements(TableInfo<Table, Object?> table) {
+  /// same format as the `hlcConverter.toSql` method.
+  String _getHlcJsonExprSql() {
+    return "printf('%015d', \"last_timestamp\") || '-' || "
+        "printf('%05x', \"counter\") || '-' || "
+        "'${crdtDb.nodeId}'";
+  }
+
+  /// Returns SQL statements to create the in-SQLite HLC view with "instead of"
+  /// trigger that will act as a function to update the HLC on the HLC state table.
+  /// This will be called from within trigger statements for the CRDT data table.
+  @override
+  List<String> generateSetupStatements() {
     return [
-      for (final op in Operation.values)
+      '''
-        wrapAsCreateTriggerStatement(
+CREATE VIEW IF NOT EXISTS "$hlcCanonicalViewName" AS
-          operation: op,
+SELECT CAST(NULL AS TEXT) AS "user_id"
-          tableName: table.actualTableName,
+WHERE FALSE;
-          triggerName: _getTriggerName(table, op),
+
-          innerSql: _getInsertStatement(
+CREATE TRIGGER IF NOT EXISTS "increment_canonical_hlc"
-            table: table,
+INSTEAD OF INSERT ON "$hlcCanonicalViewName"
-            operation: op,
+BEGIN
-          ),
+  SELECT RAISE(ABORT, 'Clock drift exceeds 1 minute maximum.')
-        ),
+  FROM "$hlcStateTableName"
+  WHERE "user_id" = NEW."user_id"
+    AND "last_timestamp" > (unixepoch('now','subsec') + 60) * 1000;
+
+  SELECT RAISE(ABORT, 'Timestamp counter overflow: 0xFFFF.')
+  FROM "$hlcStateTableName"
+  WHERE "user_id" = NEW."user_id"
+    AND "counter" = 0xFFFF;
+
+  INSERT INTO "$hlcStateTableName" ("user_id", "last_timestamp", "counter")
+  SELECT NEW."user_id", ts, 0
+  FROM (
+    SELECT CAST(unixepoch('now','subsec') * 1000 AS INTEGER) AS ts
+  )
+  WHERE TRUE
+  ON CONFLICT ("user_id") DO UPDATE SET
+    "last_timestamp" = MAX("last_timestamp", EXCLUDED."last_timestamp"),
+    "counter" = (
+      CASE
+        WHEN EXCLUDED."last_timestamp" > "last_timestamp" THEN 0
+        ELSE "counter" + 1
+      END
+    );
+END;''',
     ];
   }
 }
 
-/// Provides the CRDT control trigger statements for the SQLite3 database.
-class Sqlite3OfflineSyncTriggers extends OfflineSyncTriggers {
-  /// Creates a new instance of [Sqlite3OfflineSyncTriggers].
-  Sqlite3OfflineSyncTriggers(super.crdtDb, [super.conflictingBias]);
-
-  @override
-  String wrapAsCreateTriggerStatement({
-    required Operation operation,
-    required String tableName,
-    required String triggerName,
-    required String innerSql,
-  }) =>
-      '''
-CREATE TRIGGER IF NOT EXISTS "$triggerName"
-AFTER ${operation.name.toUpperCase()} ON "$tableName"
-WHEN (
-  SELECT MAX("crdt_triggers_on")
-  FROM "__crdt_control"
-  WHERE ("user_id" = '${crdtDb.userId}')
-    AND ("node_id" = '${crdtDb.nodeId}')
-) = TRUE
-BEGIN
-$innerSql
-END;''';
-}
-
 /// 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/hlc/converter.dart

@@ -11,7 +11,7 @@ class _HlcConverter extends TypeConverter<Hlc, String> {
   Hlc fromSql(String fromDb) => Hlc.parse(fromDb);
 
   @override
-  String toSql(Hlc value) => value.toJson();
+  String toSql(Hlc value) => value.toJson(unixTimestamp: true);
 }
 
 /// Converter for [Hlc] to store it in the database as string.

lib/src/hlc/stateful.dart

@@ -1,47 +0,0 @@
-import 'hlc.dart';
-
-/// A [Hlc] instance that is used to generate unique timestamps for CRDT operations.
-/// Will keep track of the HLC during the lifetime of the application.
-class StatefulHlc {
-  StatefulHlc._(this.userId, this.nodeId) {
-    lastHlc = _lastUserIdHlc[userId]?.copyWith(nodeId: nodeId) ?? Hlc.zero(nodeId);
-  }
-
-  static final Map<String, Hlc> _lastUserIdHlc = {};
-  static final Map<String, StatefulHlc> _instances = {};
-
-  /// Initialize the HLC for the provided [userId].
-  ///
-  /// This method allows a pre-initialization of the HLC for the node from the
-  /// last stored HLC for the user - regardless of the node ID. If no [initialHlc]
-  /// is provided, a zero HLC will be used.
-  static void initialize(String userId, Hlc initialHlc) {
-    if (_lastUserIdHlc.containsKey(userId)) return;
-    _lastUserIdHlc[userId] = initialHlc;
-  }
-
-  /// Returns a cached instance of [StatefulHlc] for the provided [nodeId].
-  static StatefulHlc cached(String userId, String nodeId) {
-    return _instances.putIfAbsent(nodeId, () => StatefulHlc._(userId, nodeId));
-  }
-
-  /// The user ID for the CRDT system.
-  final String userId;
-
-  /// The node ID for the CRDT system.
-  final String nodeId;
-
-  /// The current HLC timestamp for the user. Will be incremented on each operation.
-  late Hlc lastHlc;
-
-  /// Returns the next HLC timestamp for the current node.
-  Hlc increment() {
-    lastHlc = lastHlc.increment();
-    return lastHlc;
-  }
-
-  /// Merges another [Hlc] instance into the current one.
-  void merge(Hlc other) {
-    lastHlc = lastHlc.merge(other);
-  }
-}

lib/src/migrator.dart

@@ -65,6 +65,13 @@ class OfflineSyncMigrator extends Migrator {
   /// The CRDT execution instance.
   late final crdt = OfflineSyncCrdt(crdtDb, userId: userId, nodeId: nodeId);
 
+  /// The trigger creator for the CRDT system.
+  late final triggerCreator = switch (database.executor.dialect) {
+    SqlDialect.sqlite => Sqlite3OfflineSyncTriggers(crdtDb, conflictingBias),
+    // TODO: Implement the triggers for the Postgres dialect.
+    _ => throw UnsupportedError('Unsupported dialect: ${database.executor.dialect}'),
+  };
+
   /// Migrations for tables that are part of the CRDT system.
   final pendingMigrations = <String, CrdtTableMigration>{};
 
@@ -112,6 +119,10 @@ class OfflineSyncMigrator extends Migrator {
       // TODO: Alter all FK constraints to be deferrable initially deferred.
     }
 
+    for (final statement in triggerCreator.generateSetupStatements()) {
+      await database.customStatement(statement);
+    }
+
     for (final table in synchronizedTables) {
       await _createTriggers(table);
     }
@@ -130,11 +141,6 @@ class OfflineSyncMigrator extends Migrator {
   }
 
   Future<void> _createTriggers(TableInfo table) async {
-    final triggerCreator = switch (database.executor.dialect) {
-      SqlDialect.sqlite => Sqlite3OfflineSyncTriggers(crdtDb, conflictingBias),
-      // TODO: Implement the triggers for the Postgres dialect.
-      _ => throw UnsupportedError('Unsupported dialect: ${database.executor.dialect}'),
-    };
     final triggers = triggerCreator.generateCreateTriggerStatements(table);
     for (final trigger in triggers) {
       await database.customStatement(trigger);

lib/src/utils/sql_builder.dart

@@ -23,6 +23,9 @@ class CrdtDataSqlBuilder {
   /// Name of the CRDT data table.
   String get crdtDataTableName => crdtDb.crdtDataTable.actualTableName;
 
+  /// Name of the HLC state table.
+  String get hlcStateTableName => crdtDb.crdtHlcStateTable.actualTableName;
+
   /// The list of quoted column names in the CRDT data table.
   Iterable<String> get crdtDataAllColumns => crdtDb.crdtDataTable.$columns.quotedNames;
 

test/crdt/crdt_get_canonical_time_test.dart

@@ -0,0 +1,72 @@
+import 'package:drift/drift.dart' hide isNotNull, isNull;
+import 'package:drift_offline_sync/drift_offline_sync.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(await crdt.getCanonicalTime(), 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(await crdt.getCanonicalTime(), equals(createdHlc));
+    });
+  });
+
+  group('Given CRDT for a database with data and a new data entry', () {
+    late OfflineSyncCrdt crdt;
+    late Hlc firstCreatedHlc;
+
+    setUp(() async {
+      await database.managers.todosTable.create(
+        (t) => t(content: 'test', targetDate: Value(DateTime.now())),
+      );
+
+      (crdt, _, _) = database.crdtContext;
+
+      firstCreatedHlc = await crdt.getCanonicalTime();
+
+      await database.managers.todosTable.create(
+        (t) => t(content: 'test2', targetDate: Value(DateTime.now())),
+      );
+    });
+
+    test(
+      'when getting the canonical time '
+      'then the canonical time is increased since the first data entry was created.',
+      () async {
+        final canonicalTime = await crdt.getCanonicalTime();
+        expect(canonicalTime, greaterThan(firstCreatedHlc));
+      },
+    );
+  });
+}

test/crdt/crdt_merge_canonical_time_test.dart

@@ -0,0 +1,105 @@
+import 'package:drift_offline_sync/drift_offline_sync.dart';
+import 'package:test/test.dart';
+
+import '../utils/crdt_context.dart';
+import '../utils/database.dart';
+
+void main() {
+  group('Given CRDT with an existing canonical time', () {
+    late OfflineSyncCrdt crdt;
+    late CrdtDatabase crdtDb;
+    late Hlc initialHlc;
+
+    const otherNodeId = 'other';
+    final initialDateTime = DateTime.now().toUtc();
+
+    setUp(() async {
+      (crdt, crdtDb, _) = database.crdtContext;
+
+      await crdtDb.managers.crdtHlcStateTable.create(
+        (t) => t(
+          userId: crdt.userId,
+          lastTimestamp: initialDateTime.millisecondsSinceEpoch,
+          counter: 17,
+        ),
+      );
+
+      initialHlc = await crdt.getCanonicalTime();
+    });
+
+    test('when merging an HLC with higher time and counter '
+        'then the canonical time has the new HLC timestamp and counter.', () async {
+      final newHlc = Hlc(initialDateTime.advance(), 18, otherNodeId);
+
+      await crdt.mergeCanonicalTime(newHlc);
+      final canonicalTime = await crdt.getCanonicalTime();
+
+      expect(canonicalTime, equals(newHlc.copyWith(nodeId: crdt.nodeId)));
+    });
+
+    test('when merging an HLC with higher time and lower counter '
+        'then the canonical time has the new HLC timestamp and counter.', () async {
+      final newHlc = Hlc(initialDateTime.advance(), 0, otherNodeId);
+
+      await crdt.mergeCanonicalTime(newHlc);
+      final canonicalTime = await crdt.getCanonicalTime();
+
+      expect(canonicalTime, equals(newHlc.copyWith(nodeId: crdt.nodeId)));
+    });
+
+    test('when merging an HLC with lower time and counter '
+        'then the canonical time is the initial HLC.', () async {
+      final newHlc = Hlc(initialDateTime.retreat(), 0, otherNodeId);
+
+      await crdt.mergeCanonicalTime(newHlc);
+      final canonicalTime = await crdt.getCanonicalTime();
+
+      expect(canonicalTime, equals(initialHlc));
+    });
+
+    test('when merging an HLC with lower time and higher counter '
+        'then the canonical time is the initial HLC.', () async {
+      final newHlc = Hlc(initialDateTime.retreat(), 18, otherNodeId);
+
+      await crdt.mergeCanonicalTime(newHlc);
+      final canonicalTime = await crdt.getCanonicalTime();
+
+      expect(canonicalTime, equals(initialHlc));
+    });
+
+    test('when merging an HLC with same time and higher counter '
+        'then the canonical time has the higher counter.', () async {
+      final newHlc = Hlc(initialDateTime, 18, otherNodeId);
+
+      await crdt.mergeCanonicalTime(newHlc);
+      final canonicalTime = await crdt.getCanonicalTime();
+
+      expect(canonicalTime, equals(newHlc.copyWith(nodeId: crdt.nodeId)));
+    });
+
+    test('when merging an HLC with same time and lower counter '
+        'then the canonical time is the initial HLC.', () async {
+      final newHlc = Hlc(initialDateTime, 0, otherNodeId);
+
+      await crdt.mergeCanonicalTime(newHlc);
+      final canonicalTime = await crdt.getCanonicalTime();
+
+      expect(canonicalTime, equals(initialHlc));
+    });
+
+    test('when merging an HLC with same time and counter '
+        'then the canonical time is the initial HLC.', () async {
+      final newHlc = Hlc(initialDateTime, 17, otherNodeId);
+
+      await crdt.mergeCanonicalTime(newHlc);
+      final canonicalTime = await crdt.getCanonicalTime();
+
+      expect(canonicalTime, equals(initialHlc));
+    });
+  });
+}
+
+extension on DateTime {
+  DateTime advance() => add(const Duration(milliseconds: 1));
+  DateTime retreat() => subtract(const Duration(milliseconds: 1));
+}

test/database_setup_test.dart

@@ -1,110 +0,0 @@
-import 'package:drift/drift.dart';
-import 'package:drift/native.dart';
-import 'package:drift_offline_sync/drift_offline_sync.dart';
-import 'package:drift_offline_sync/src/database/triggers.dart';
-import 'package:test/test.dart';
-
-void main() {
-  group('Given an executor with the HLC function registered', () {
-    late NativeDatabase executor;
-
-    setUp(() async {
-      executor = NativeDatabase.memory(setup: registerHlcFunction);
-      await executor.ensureOpen(_TestQueryExecutorUser());
-    });
-
-    tearDown(() async {
-      await executor.close();
-    });
-
-    test('when calling the HLC function with valid arguments '
-        'then it returns a valid HLC timestamp.', () async {
-      final result = await executor.runSelect(
-        'SELECT $nextHlcFunction(?, ?) as hlc_timestamp;',
-        ['user_id', 'node_id'],
-      );
-
-      final hlcTimestamp = result.first['hlc_timestamp'] as String?;
-      expect(hlcTimestamp, isA<String>());
-      expect(hlcConverter.fromSql(hlcTimestamp!), isA<Hlc>());
-    });
-
-    test('when calling the HLC function with no arguments '
-        'then it throws a SqliteException due to the MissingArgumentsError.', () async {
-      expect(
-        () => executor.runSelect(
-          'SELECT $nextHlcFunction() as hlc_timestamp;',
-          [],
-        ),
-        throwsA(
-          isA<SqliteException>().having(
-            (e) => e.message,
-            'message',
-            contains('MissingArgumentsError'),
-          ),
-        ),
-      );
-    });
-
-    test('when calling the HLC function with only the first argument '
-        'then it throws a SqliteException due to the MissingNodeIdError.', () async {
-      expect(
-        () => executor.runSelect(
-          'SELECT $nextHlcFunction(?) as hlc_timestamp;',
-          ['user_id'],
-        ),
-        throwsA(
-          isA<SqliteException>().having(
-            (e) => e.message,
-            'message',
-            contains('MissingNodeIdError'),
-          ),
-        ),
-      );
-    });
-
-    test('when calling the HLC function with only the second argument '
-        'then it throws a SqliteException due to the MissingUserIdError.', () async {
-      expect(
-        () => executor.runSelect(
-          'SELECT $nextHlcFunction(?, ?) as hlc_timestamp;',
-          [null, 'user_id'],
-        ),
-        throwsA(
-          isA<SqliteException>().having(
-            (e) => e.message,
-            'message',
-            contains('MissingUserIdError'),
-          ),
-        ),
-      );
-    });
-
-    test('when calling the HLC function with more than two arguments '
-        'then it throws a SqliteException due to the TooManyArgumentsError.', () async {
-      expect(
-        () => executor.runSelect(
-          'SELECT $nextHlcFunction(?, ?, ?) as hlc_timestamp;',
-          ['user_id', 'node_id', 'extra_argument'],
-        ),
-        throwsA(
-          isA<SqliteException>().having(
-            (e) => e.message,
-            'message',
-            contains('TooManyArgumentsError'),
-          ),
-        ),
-      );
-    });
-  });
-}
-
-class _TestQueryExecutorUser implements QueryExecutorUser {
-  @override
-  int get schemaVersion => 0;
-
-  @override
-  Future<void> beforeOpen(QueryExecutor executor, OpeningDetails details) async {
-    return;
-  }
-}

test/hlc/hlc_converter_test.dart

@@ -7,10 +7,13 @@ final _dateTime = DateTime.parse(_isoTime);
 void main() {
   test('Given an HLC and the HlcConverter '
       'when converting to the database format '
-      'then it returns string with HLC JSON format.', () {
+      'then it returns string with padded unix timestamp.', () {
     final hlc = Hlc(_dateTime, 0x17, 'abc');
 
-    expect(hlcConverter.toSql(hlc), '${_dateTime.toIso8601String()}-0017-abc');
+    expect(
+      hlcConverter.toSql(hlc),
+      '${_dateTime.millisecondsSinceEpoch.toString().padLeft(15, '0')}-0017-abc',
+    );
   });
 
   test('Given an HLC and the HlcConverter '

test/hlc/hlc_increment_test.dart

@@ -54,7 +54,7 @@ void main() {
   });
 
   group('Given an HLC with canonical time more than one minute ahead of wall time', () {
-    final hlc = Hlc(_dateTime.advance(const Duration(minutes: 2)), 0, 'abc');
+    final hlc = Hlc(_dateTime.add(const Duration(minutes: 1, seconds: 5)), 0, 'abc');
     final wallTime = _dateTime;
 
     test('when incrementing then ClockDriftException is thrown.', () {
@@ -77,9 +77,6 @@ void main() {
 }
 
 extension on DateTime {
-  DateTime advance([Duration duration = const Duration(milliseconds: 1)]) =>
+  DateTime advance() => add(const Duration(milliseconds: 1));
-      add(duration);
+  DateTime retreat() => subtract(const Duration(milliseconds: 1));
-
-  DateTime retreat([Duration duration = const Duration(milliseconds: 1)]) =>
-      subtract(duration);
 }

test/stateful_hlc_initialize_test.dart

@@ -1,98 +0,0 @@
-import 'package:drift_offline_sync/drift_offline_sync.dart';
-import 'package:drift_offline_sync/src/hlc/stateful.dart';
-import 'package:test/test.dart';
-
-import 'utils/crdt_context.dart';
-import 'utils/database.dart';
-import 'utils/executor.dart';
-import 'utils/user.dart';
-
-void main() {
-  group(
-    'Given a database opened with a CRDT control table having initial HLC from same user and node',
-    () {
-      late Hlc initialHlc;
-
-      setUp(() async {
-        initialHlc = Hlc.now(testNodeId);
-        await testDatabaseFixture(initialHlc: initialHlc);
-      });
-
-      test('when retrieving the cached HLC '
-          'then the HLC for the node is the initial HLC.', () async {
-        expect(StatefulHlc.cached(testUserId, testNodeId).lastHlc, equals(initialHlc));
-      });
-    },
-  );
-
-  group(
-    'Given a database opened with a CRDT control table having initial HLC from same user and different node',
-    () {
-      late Hlc initialHlc;
-
-      setUp(() async {
-        initialHlc = Hlc.now('other_node_id');
-        await testDatabaseFixture(initialHlc: initialHlc);
-      });
-
-      test(
-        'when retrieving the cached HLC '
-        'then the HLC for the node is the initial HLC for the requested node.',
-        () async {
-          final expectedHlc = initialHlc.copyWith(nodeId: testNodeId);
-          expect(
-            StatefulHlc.cached(testUserId, testNodeId).lastHlc,
-            equals(expectedHlc),
-          );
-        },
-      );
-    },
-  );
-
-  group(
-    'Given a database opened with a CRDT control table having initial HLC from different user and node',
-    () {
-      late Hlc initialHlc;
-
-      setUp(() async {
-        initialHlc = Hlc.now('other_node_id');
-        await testDatabaseFixture(initialHlc: initialHlc);
-      });
-
-      test('when retrieving the cached HLC '
-          'then the HLC for the node is zero for the requested node.', () async {
-        final expectedHlc = Hlc.zero(testNodeId);
-        expect(
-          StatefulHlc.cached('other_user_id', testNodeId).lastHlc,
-          equals(expectedHlc),
-        );
-      });
-    },
-  );
-}
-
-Future<TodoDb> testDatabaseFixture({
-  required Hlc initialHlc,
-  String? userId,
-}) async {
-  setTestFileExecutor();
-  final database = TodoDb(testExecutor);
-
-  final (crdt, crdtDb, nodeId) = database.crdtContext;
-
-  await crdtDb.managers.crdtDataTable.create(
-    (t) => t(
-      userId: userId ?? crdt.userId,
-      tblName: 'todos',
-      columnName: 'content',
-      rowId: '1',
-      hlcTimestamp: initialHlc,
-    ),
-  );
-
-  await database.close();
-  await testExecutor.close();
-
-  setTestFileExecutor();
-  return TodoDb(testExecutor);
-}