fix: Return the userId to the CRDT data table

lib/src/hlc/stateful.dart

@@ -3,15 +3,31 @@ import 'package:crdt/crdt.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.nodeId) : lastHlc = Hlc.now(nodeId);
+  StatefulHlc._(this.userId, this.nodeId) {
+    lastHlc = _lastUserIdHlc[userId]?.apply(nodeId: nodeId) ?? Hlc.zero(nodeId);
+  }
 
+  static final Map<String, Hlc> _lastUserIdHlc = {};
   static final Map<String, StatefulHlc> _instances = {};
 
-  /// Returns a cached instance of [StatefulHlc] for the provided [nodeId].
-  static StatefulHlc cached(String nodeId) {
-    return _instances.putIfAbsent(nodeId, () => StatefulHlc._(nodeId));
+  /// 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;
 

lib/src/migrator.dart

@@ -79,8 +79,8 @@ class OfflineSyncMigrator extends Migrator {
       final crdtTable = crdtDb.crdtDataTable;
       await database.customStatement(
         'ALTER TABLE ${crdtTable.actualTableName} '
-        'PARTITION BY RANGE (${crdtTable.tblName.name});',
-        // 'PARTITION BY RANGE (${crdtTable.userId.name}, ${crdtTable.tblName.name});',
+        // TODO: Fix this partitioning, since RANGE can not be used for discrete values.
+        'PARTITION BY RANGE (${crdtTable.userId.name}, ${crdtTable.tblName.name});',
       );
     }
 

lib/src/setup.dart

@@ -2,6 +2,7 @@ import 'package:drift/native.dart';
 
 import '/src/hlc/stateful.dart';
 import '/src/triggers.dart';
+import 'tables/data.dart';
 
 /// The database setup function to register the HLC function.
 ///
@@ -32,6 +33,18 @@ DatabaseSetup get registerHlcFunction {
       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.
+    }
   };
 }
 

lib/src/tables/data.dart

@@ -17,15 +17,14 @@ export '../hlc/converter.dart';
 /// the client will have the `user_id` column with a single value, while the server
 /// will have all users' data in the same table. To avoid performance issues, on a
 /// Postgres server, the `user_id` and `table_name` columns are partition keys.
-///
 @TableIndex(
   name: 'idx_crdt_data_tbl_name_column_name_row_id',
-  columns: {#tblName, #columnName, #rowId},
+  columns: {#userId, #tblName, #columnName, #rowId},
   unique: true,
 )
 @TableIndex(
   name: 'idx_crdt_data_hlc_timestamp',
-  columns: {#hlcTimestamp},
+  columns: {#userId, #hlcTimestamp},
 )
 @DataClassName('CrdtDataEntry')
 class CrdtDataTable extends Table {
@@ -35,14 +34,8 @@ class CrdtDataTable extends Table {
   @override
   bool get isStrict => true;
 
-  // MAYBE: It seems it is not possible to have the userId column here without
-  // creating hard requirements on how each table should relate to the userId.
-  // Either all tables would have to have this column, or users would have to
-  // supply a query to join the userId with the changed table. Either way, it
-  // makes the module a lot less plug-and-play than it should, so it might be
-  // better to not have it here and accept the query performance penalty.
-  // /// Identifier for the user or client that owns the data.
-  // late final userId = text()();
+  /// Identifier for the user or client that owns the data.
+  late final userId = text()();
 
   /// Name of the table that data belongs to.
   late final tblName = text().named('table_name')();
@@ -60,5 +53,17 @@ class CrdtDataTable extends Table {
   late final rawValue = sqliteAny().nullable()();
 
   @override
-  Set<Column> get primaryKey => {/*userId,*/ tblName, columnName, rowId};
+  Set<Column> get primaryKey => {userId, tblName, columnName, rowId};
+
+  /// Query to get the last HLC timestamp for each user.
+  ///
+  /// Declared here to ensure the query is synchronized with the table.
+  static String getLastHlcTimestampQuery() {
+    return '''
+    SELECT user_id,
+           max(hlc_timestamp) as last_hlc_timestamp
+    FROM __crdt_data
+    GROUP BY 1
+  ''';
+  }
 }

lib/src/triggers.dart

@@ -95,13 +95,14 @@ abstract class OfflineSyncTriggers {
     ${_crdtDataAllColumns.join(', ')}
   )
   SELECT
+    '${migrator.userId}' AS "user_id",
     '$tableName' AS "table_name",
     column_name,
     $uniqueRowId AS "row_id",
     "hlc_timestamp",
     raw_value
   FROM (
-    SELECT $nextHlcFunction('${migrator.nodeId}') AS "hlc_timestamp"
+    SELECT $nextHlcFunction('${migrator.userId}', '${migrator.nodeId}') AS "hlc_timestamp"
   ), (
 ${columnInserts.join('\n    UNION ALL\n')}
   )$whereClause;''';

test/triggers/trigger_generation_test.dart

@@ -38,16 +38,17 @@ CREATE TRIGGER IF NOT EXISTS __crdt__todos__insert
 AFTER INSERT ON "todos"
 BEGIN
   INSERT OR REPLACE INTO __crdt_data (
-    "table_name", "column_name", "row_id", "hlc_timestamp", "raw_value"
+    "user_id", "table_name", "column_name", "row_id", "hlc_timestamp", "raw_value"
   )
   SELECT
+    '${migrator.userId}' AS "user_id",
     'todos' AS "table_name",
     column_name,
     NEW."id" AS "row_id",
     "hlc_timestamp",
     raw_value
   FROM (
-    SELECT $nextHlcFunction('${migrator.nodeId}') AS "hlc_timestamp"
+    SELECT $nextHlcFunction('${migrator.userId}', '${migrator.nodeId}') AS "hlc_timestamp"
   ), (
     SELECT
       'id' AS "column_name",
@@ -88,17 +89,17 @@ CREATE TRIGGER IF NOT EXISTS __crdt__todos__update
 AFTER UPDATE ON "todos"
 BEGIN
   INSERT OR REPLACE INTO __crdt_data (
-    "table_name", "column_name", "row_id", "hlc_timestamp", "raw_value"
+    "user_id", "table_name", "column_name", "row_id", "hlc_timestamp", "raw_value"
   )
   SELECT
+    '${migrator.userId}' AS "user_id",
     'todos' AS "table_name",
     column_name,
     OLD."id" AS "row_id",
     "hlc_timestamp",
-    raw_value,
-    value_changed
+    raw_value
   FROM (
-    SELECT $nextHlcFunction('${migrator.nodeId}') AS "hlc_timestamp"
+    SELECT $nextHlcFunction('${migrator.userId}', '${migrator.nodeId}') AS "hlc_timestamp"
   ), (
     SELECT
       'id' AS "column_name",
@@ -146,16 +147,17 @@ CREATE TRIGGER IF NOT EXISTS __crdt__todos__delete
 AFTER DELETE ON "todos"
 BEGIN
   INSERT OR REPLACE INTO __crdt_data (
-    "table_name", "column_name", "row_id", "hlc_timestamp", "raw_value"
+    "user_id", "table_name", "column_name", "row_id", "hlc_timestamp", "raw_value"
   )
   SELECT
+    '${migrator.userId}' AS "user_id",
     'todos' AS "table_name",
     column_name,
     OLD."id" AS "row_id",
     "hlc_timestamp",
     raw_value
   FROM (
-    SELECT $nextHlcFunction('${migrator.nodeId}') AS "hlc_timestamp"
+    SELECT $nextHlcFunction('${migrator.userId}', '${migrator.nodeId}') AS "hlc_timestamp"
   ), (
     SELECT
       '__crdt_is_deleted' AS "column_name",