chore: Reorganize the package

lib/drift_offline_first.dart

@@ -3,13 +3,13 @@
 /// More docs go here.
 library;
 
-export 'src/crdt.dart';
+export 'src/crdt/base.dart';
-export 'src/crdt/extensions/convert.dart';
+export 'src/crdt/crdt.dart';
-export 'src/crdt/extensions/entry.dart';
+export 'src/database/database.dart';
-export 'src/crdt/extensions/extract.dart';
+export 'src/database/extensions/convert.dart';
-export 'src/crdt/extensions/sync.dart';
+export 'src/database/extensions/entry.dart';
-export 'src/crdt_base.dart';
+export 'src/database/extensions/extract.dart';
-export 'src/database.dart';
+export 'src/database/extensions/sync.dart';
+export 'src/database/setup.dart';
 export 'src/hlc/converter.dart';
 export 'src/migrator.dart';
-export 'src/setup.dart';

lib/src/crdt/base.dart

@@ -0,0 +1,106 @@
+import 'dart:async';
+
+import 'package:crdt/crdt.dart';
+import 'package:meta/meta.dart';
+
+import '../database/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/crdt/crdt.dart

@@ -0,0 +1,135 @@
+import 'package:crdt/crdt.dart';
+import 'package:drift/drift.dart';
+
+import '../database/database.dart';
+import 'base.dart';
+
+/// A CRDT implementation for offline-first synchronization.
+class OfflineSyncCrdt extends CrdtBase {
+  /// Creates a new instance of [OfflineSyncCrdt].
+  ///
+  /// The [nodeId] must be unique across the system.
+  OfflineSyncCrdt(this._db, {required super.userId, required super.nodeId});
+
+  final CrdtDatabase _db;
+
+  @override
+  // TODO: Remove this method lately if it is not needed - as it seems it will.
+  Future<Hlc?> getLastModified({String? onlyNodeId, String? exceptNodeId}) async {
+    if (onlyNodeId != null && exceptNodeId != null) {
+      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();
+
+    return rowWithMaxLastSyncHlc?.hlcTimestamp;
+  }
+
+  @override
+  Future<Iterable<CrdtDataEntry>> getChangeset({
+    String? onlyNodeId,
+    String? exceptNodeId,
+    Hlc? modifiedAfter,
+  }) async {
+    assert(onlyNodeId == null || exceptNodeId == null);
+
+    // Modified times use the local node id
+    modifiedAfter = modifiedAfter?.apply(nodeId: nodeId);
+
+    final query = _db.managers.crdtDataTable.filter((o) {
+      Expression<bool> condition = const Constant(true);
+
+      if (onlyNodeId != null) {
+        condition &= o.hlcTimestamp.column.contains(onlyNodeId);
+      }
+      if (exceptNodeId != null) {
+        condition &= o.hlcTimestamp.column.contains(exceptNodeId).not();
+      }
+      if (modifiedAfter != null) {
+        condition &= o.hlcTimestamp.column.isBiggerThanValue(modifiedAfter.toString());
+      }
+      return condition;
+    });
+
+    return query.get();
+  }
+
+  @override
+  Future<void> saveChangeset(
+    Iterable<CrdtDataEntry> changeset,
+    Iterable<Hlc> receivedHlcs,
+  ) async {
+    await _db.transaction(() async {
+      await _saveChangeset(changeset);
+      await _applyCompensationRules();
+      await _saveMergedHlcs(receivedHlcs);
+    });
+  }
+
+  /// 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 {
+    // 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/database/database.dart

@@ -0,0 +1,73 @@
+import 'package:drift/drift.dart';
+
+import '../utils/sql_builder.dart';
+import 'tables/compensation.dart';
+import 'tables/control.dart';
+import 'tables/data.dart';
+import 'tables/merge.dart';
+
+part 'database.g.dart';
+
+/// The main database class for CRDT synchronization.
+@DriftDatabase(
+  tables: [
+    CrdtDataTable,
+    CrdtControlTable,
+    CrdtCompensationTable,
+    CrdtMergeHlcTable,
+  ],
+)
+class CrdtDatabase extends _$CrdtDatabase {
+  /// Creates a new instance of [CrdtDatabase] with the provided [executor].
+  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;
+
+  /// Not actually used, but required by the generated code. Will be ignored because
+  /// the [CrdtDatabase] always receives an already opened executor. No migrations
+  /// 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<T extends Insertable>([String? tableName]) {
+    if (tableName == null && T == dynamic) {
+      throw ArgumentError('Table name is required when T is dynamic.');
+    }
+    return firstWhere(
+      (t) =>
+          (tableName == null && t is TableInfo<Table, T>) ||
+          t.actualTableName == tableName,
+      orElse: () => throw ArgumentError(
+        'Table ${tableName != null ? '"$tableName"' : 'for type "$T"'} not '
+        'found in synchronized tables.',
+      ),
+    );
+  }
+}

lib/src/database/extensions/convert.dart

@@ -0,0 +1,59 @@
+import 'package:drift/drift.dart';
+
+import '../../hlc/converter.dart';
+import '../database.dart';
+
+/// Extensions for [CrdtDatabase] to convert data to [CrdtDataEntry].
+extension CrdtDatabaseConvertExtensions on CrdtDatabase {
+  /// 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.
+  ///
+  /// Returns a list of [CrdtDataEntry].
+  Iterable<CrdtDataEntry> convertToCrdtDataEntry<T extends Insertable>(
+    T data,
+    Hlc hlcTimestamp,
+  ) {
+    final json = data.toJsonForDatabase(_generationContext);
+    final tableInfo = synchronizedTables.whereType<TableInfo<dynamic, T>>().single;
+    final rowIdValues = [for (final c in tableInfo.$primaryKey) json[c.$name]];
+
+    return tableInfo.$columns.map(
+      (c) => CrdtDataEntry(
+        userId: userId,
+        tblName: tableInfo.actualTableName,
+        columnName: c.$name,
+        rowId: rowIdValues.join(sqlBuilder.rowIdSeparator),
+        rawValue: json[c.$name]?.toDriftAny(),
+        hlcTimestamp: hlcTimestamp,
+      ),
+    );
+  }
+
+  /// Gets a [GenerationContext] for the database.
+  GenerationContext get _generationContext => GenerationContext.fromDb(this);
+}
+
+extension on Insertable {
+  Map<String, Object?> toJsonForDatabase(GenerationContext context) {
+    final expressions = toColumns(false);
+    return {
+      for (final e in expressions.entries) e.key: e.value.extractValue(context),
+    };
+  }
+}
+
+extension on Expression {
+  Object? extractValue(GenerationContext context) {
+    final exp = this;
+    if (exp is! Variable) {
+      throw ArgumentError.value(exp, 'exp', 'Expected a Variable');
+    }
+    return exp.mapToSimpleValue(context);
+  }
+}
+
+extension on Object {
+  DriftAny? toDriftAny() => DriftAny(this);
+}

lib/src/database/extensions/entry.dart

@@ -0,0 +1,25 @@
+import '../../hlc/converter.dart';
+import '../database.dart';
+
+/// Extensions for the [CrdtDataEntry] class.
+extension CrdtDataEntryUnwrapExtensions on CrdtDataEntry {
+  /// Returns the unwrapped raw value, if any.
+  Object? get unwrappedValue => rawValue?.rawSqlValue;
+}
+
+/// Extensions for an [Iterable] of [CrdtDataEntry].
+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);
+
+  /// Returns the entries with the HLC replaced by the [hlcTimestamp].
+  Iterable<CrdtDataEntry> withHlc(Hlc hlcTimestamp) =>
+      map((e) => e.copyWith(hlcTimestamp: hlcTimestamp));
+}

lib/src/database/extensions/extract.dart

@@ -0,0 +1,52 @@
+import 'package:drift/drift.dart';
+
+import '../database.dart';
+
+/// Extensions for [CrdtDatabase] to extract data from the CRDT data table.
+extension CrdtDatabaseExtractExtensions on CrdtDatabase {
+  /// Gets a single data entry from the CRDT data table.
+  Future<T?> getSingleFromCrdtData<T extends Insertable>(String rowId) async {
+    final data = await getFromCrdtData<T>([rowId]);
+    return data.firstOrNull;
+  }
+
+  /// Gets a list of data entries from the CRDT data table.
+  Future<Iterable<T>> getFromCrdtData<T extends Insertable>([
+    Iterable<String>? rowIds,
+  ]) async {
+    final tableInfo = synchronizedTables.find<T>();
+    final tableName = tableInfo.actualTableName;
+    final crdtDataEntries = await managers.crdtDataTable.filter(
+      (o) {
+        var condition = o.tblName.equals(tableName);
+        if (rowIds != null) {
+          condition &= o.rowId.isIn(rowIds);
+        }
+        return condition;
+      },
+    ).get();
+
+    return tableInfo.fromCrdtDataEntries<T>(crdtDataEntries);
+  }
+}
+
+/// Extensions for [TableInfo] to convert [CrdtDataEntry] to [Insertable].
+extension ConvertCrdtDataEntry on TableInfo {
+  /// Converts [CrdtDataEntry] to [Insertable].
+  Future<Iterable<T>> fromCrdtDataEntries<T extends Insertable>(
+    Iterable<CrdtDataEntry> entries,
+  ) async {
+    final foundRowIds = entries.map((e) => e.rowId).toSet();
+    return [
+      for (final rowId in foundRowIds)
+        await map({
+              for (final c in $columns)
+                c.$name: entries
+                    .firstWhere((e) => e.rowId == rowId && e.columnName == c.$name)
+                    .rawValue
+                    ?.rawSqlValue,
+            })
+            as T,
+    ];
+  }
+}

lib/src/database/extensions/sync.dart

@@ -0,0 +1,136 @@
+import 'package:crdt/crdt.dart';
+import 'package:drift/drift.dart';
+
+import '../../utils/sql_builder.dart';
+import '../database.dart';
+import 'transactions.dart';
+
+/// Extensions for [CrdtDatabase] to sync data to the target tables.
+extension CrdtDatabaseSyncExtensions on CrdtDatabase {
+  /// Upserts rows in the target table from the CRDT data table.
+  ///
+  /// This is the inverse operation of what CRDT triggers do. It reads from the
+  /// CRDT data table and reconstructs rows to insert/update in the target table.
+  /// Handles deletions and fields that are not changed in the filter criteria.
+  ///
+  /// Parameters:
+  /// - [rowIdsPerTable]: Map of table names to row IDs to sync. If not provided,
+  ///   all tables will be synchronized. Passing null for a table will synchronize
+  ///   all rows in that table.
+  /// - [sinceHlc]: Optional HLC timestamp to filter by. The filter is inclusive,
+  ///   meaning that rows with the exact same HLC timestamp will also be included.
+  Future<void> syncCrdtDataToTables({
+    Map<String, Iterable<String>?>? rowIdsPerTable,
+    Hlc? sinceHlc,
+  }) async {
+    return transactionWithDeferredConstraints(() async {
+      final tableNames = rowIdsPerTable?.keys ?? synchronizedTables.tableNames;
+      for (final tableName in tableNames) {
+        await _syncCrdtDataToTable(
+          tableName: tableName,
+          rowIds: rowIdsPerTable?[tableName],
+          sinceHlc: sinceHlc,
+        );
+      }
+    });
+  }
+
+  /// Syncs the CRDT data to the target table.
+  ///
+  /// Must be called within a transaction with deferred constraints to ensure that
+  /// the foreign key constraints are not violated. The transaction is not created
+  /// by this method to allow grouping multiple table syncs into one transaction
+  /// to avoid nested transactions.
+  Future<void> _syncCrdtDataToTable({
+    required String tableName,
+    Iterable<String>? rowIds,
+    Hlc? sinceHlc,
+  }) async {
+    final deleteSql = _getDeleteFromCrdtDataSql(
+      tableName: tableName,
+      rowIds: rowIds,
+      sinceHlc: sinceHlc,
+    );
+
+    final upsertSql = _getSaveFromCrdtDataSql(
+      tableName: tableName,
+      rowIds: rowIds,
+      sinceHlc: sinceHlc,
+    );
+
+    await customStatement(deleteSql);
+    await customStatement(upsertSql);
+  }
+
+  /// Generates the SQL to delete rows in the target table from the CRDT data table.
+  String _getDeleteFromCrdtDataSql({
+    required String tableName,
+    Iterable<String>? rowIds,
+    Hlc? sinceHlc,
+  }) {
+    return '''
+    DELETE FROM "$tableName"
+    WHERE ${_getWhereClause(tableName, rowIds, sinceHlc, isDeleted: true)};''';
+  }
+
+  // TODO: Unify this with the SQL for the CRDT triggers.
+  /// Generates SQL to upsert rows in the target table from the CRDT data table.
+  String _getSaveFromCrdtDataSql({
+    required String tableName,
+    Iterable<String>? rowIds,
+    Hlc? sinceHlc,
+  }) {
+    final tableInfo = synchronizedTables.find(tableName);
+
+    final columnSelects = tableInfo.persistedColumns.map((column) {
+      final whereClause = 'WHERE c."column_name" = \'$column\'';
+      return 'MAX(c."raw_value") FILTER ($whereClause) AS "$column"';
+    });
+
+    final updateSets = tableInfo.nonPrimaryKeyColumns.map((column) {
+      final newField = 'EXCLUDED."$column"';
+      final originalField = '"$tableName"."$column"';
+      return '"$column" = COALESCE($newField, $originalField)';
+    });
+
+    final onConflict = updateSets.isEmpty
+        ? 'DO NOTHING'
+        : 'DO UPDATE\nSET ${updateSets.join(',\n    ')}';
+
+    return '''
+INSERT INTO "$tableName" (
+  ${tableInfo.persistedColumns.map((c) => '"$c"').join(', ')}
+)
+SELECT ${columnSelects.join(',\n       ')}
+FROM ${sqlBuilder.crdtDataTableName} AS c
+WHERE ${_getWhereClause(tableName, rowIds, sinceHlc)}
+GROUP BY c."row_id"
+ON CONFLICT (${tableInfo.$primaryKey.quotedNames.join(', ')})
+$onConflict;''';
+  }
+
+  String _getWhereClause(
+    String tableName,
+    Iterable<String>? rowIds,
+    Hlc? sinceHlc, {
+    bool isDeleted = false,
+  }) {
+    final whereConditions = <String>[
+      '"user_id" = \'$userId\'',
+      '"table_name" = \'$tableName\'',
+      if (rowIds != null && rowIds.isNotEmpty)
+        'row_id IN (${rowIds.map((id) => "'$id'").join(', ')})',
+      if (sinceHlc != null) '"hlc_timestamp" >= \'$sinceHlc\'',
+      if (isDeleted) '"${sqlBuilder.isDeletedColumnName}" = TRUE',
+    ];
+    return whereConditions.join(' AND ');
+  }
+}
+
+extension on TableInfo {
+  Iterable<String> get persistedColumns =>
+      $columns.where((c) => c.generatedAs == null).map((c) => c.$name);
+
+  Iterable<String> get nonPrimaryKeyColumns =>
+      persistedColumns.where((c) => !$primaryKey.map((p) => p.$name).contains(c));
+}

lib/src/database/extensions/transactions.dart

@@ -0,0 +1,36 @@
+import '../database.dart';
+
+/// Extensions for [CrdtDatabase] to execute transactions.
+extension CrdtDatabaseTransactionsExtensions on CrdtDatabase {
+  /// Executes the [operation] in a transaction with the constraints deferred.
+  ///
+  /// On Postgres, this will result in deferring the constraints for the duration
+  /// of the operation. On SQLite, this will result in disabling the foreign keys
+  /// before the operation and restoring to the previous value afterwards.
+  Future<T> transactionWithDeferredConstraints<T>(
+    Future<T> Function() operation,
+  ) async {
+    if (isPostgres) {
+      return transaction(() async {
+        await customStatement('SET CONSTRAINTS ALL DEFERRED');
+        return operation();
+      });
+    }
+
+    return _withForeignKeysDisabled(
+      () => transaction(operation),
+    );
+  }
+
+  Future<T> _withForeignKeysDisabled<T>(Future<T> Function() operation) async {
+    const pragma = 'PRAGMA foreign_keys';
+    final originalValueResult = await customSelect(pragma).getSingle();
+    final originalValue = originalValueResult.read<int>('foreign_keys');
+    try {
+      await customStatement('$pragma = OFF');
+      return await operation();
+    } finally {
+      await customStatement('$pragma = $originalValue');
+    }
+  }
+}

lib/src/database/setup.dart

@@ -0,0 +1,57 @@
+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/compensation.dart

@@ -0,0 +1,61 @@
+import 'package:drift/drift.dart';
+
+/// CRDT Compensation rules.
+///
+/// This table stores the compensation rules for each table and column.
+///
+@DataClassName('CrdtCompensationEntry')
+class CrdtCompensationTable extends Table {
+  @override
+  String get tableName => '__crdt_compensation';
+
+  /// Name of the table that data belongs to.
+  late final tblName = text().named('table_name')();
+
+  /// Name of the column this data belongs to.
+  late final columnName = text()();
+
+  /// Type of constraint this compensation rule applies to.
+  late final constraintType = textEnum<ConstraintType>()();
+
+  /// Compensation rule to preserve invariants.
+  late final compensationRule = textEnum<CompensationRule>()();
+
+  @override
+  Set<Column> get primaryKey => {tblName, columnName, constraintType};
+}
+
+/// Type of constraint this compensation rule applies to.
+enum ConstraintType {
+  /// Unique constraint.
+  unique,
+
+  /// Foreign key constraint.
+  foreignKey,
+}
+
+/// Available compensation rules for CRDT operations.
+enum CompensationRule {
+  /// No compensation rule.
+  none,
+
+  /// Handle unique key conflicts by merging values.
+  // ignore: constant_identifier_names
+  on_unique_conflict_merge,
+
+  /// Handle unique key conflicts by overwriting the existing value.
+  // ignore: constant_identifier_names
+  on_unique_conflict_overwrite,
+
+  /// Handle foreign key deletion by deleting this related row.
+  // ignore: constant_identifier_names
+  on_fk_deleted_delete_this,
+
+  /// Handle foreign key deletion by setting the relation on this row to null.
+  // ignore: constant_identifier_names
+  on_fk_deleted_set_null,
+
+  /// Handle foreign key deletion by restoring the related row.
+  // ignore: constant_identifier_names
+  on_fk_deleted_restore_related,
+}

lib/src/database/tables/control.dart

@@ -0,0 +1,37 @@
+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 control table.
+///
+/// This table stores metadata about the CRDT synchronization state for each user.
+///
+@DataClassName('CrdtControlEntry')
+class CrdtControlTable extends Table {
+  @override
+  String get tableName => '__crdt_control';
+
+  /// Identifier for the user or client.
+  late final userId = text()();
+
+  /// Node identifier for the client.
+  late final nodeId = text()();
+
+  /// Whether CRDT triggers are currently enabled for this user.
+  late final crdtTriggersOn = boolean().withDefault(const Constant(true))();
+
+  /// The schema version of the CRDT data.
+  late final schemaVersion = integer()();
+
+  /// The last HLC timestamp when the user synchronized.
+  late final lastSyncHlc = text().nullable().map(nullableHlcConverter)();
+
+  /// The last HLC timestamp when merged changes were applied.
+  late final lastApplyHlc = text().nullable().map(nullableHlcConverter)();
+
+  @override
+  Set<Column> get primaryKey => {userId, nodeId};
+}

lib/src/database/tables/data.dart

@@ -0,0 +1,69 @@
+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 data table.
+///
+/// This table stores the actual data encoded as JSON strings and all CRDT required
+/// metadata to perform conflict resolution and synchronization. Schema is designed
+/// to be generic enough to support various data types and structures. Stores the
+/// data for all tables that are registered for CRDT synchronization, simplifying
+/// the process of conflict resolution and data merging.
+///
+/// This same schema will be used on a client and server, with the difference that
+/// 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: {#userId, #tblName, #columnName, #rowId},
+  unique: true,
+)
+@TableIndex(
+  name: 'idx_crdt_data_hlc_timestamp',
+  columns: {#userId, #hlcTimestamp},
+)
+@DataClassName('CrdtDataEntry')
+class CrdtDataTable extends Table {
+  @override
+  String get tableName => '__crdt_data';
+
+  @override
+  bool get isStrict => true;
+
+  /// 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')();
+
+  /// Name of the column this data belongs to.
+  late final columnName = text()();
+
+  /// Unique identifier for the row in the table.
+  late final rowId = text()();
+
+  /// Hybrid Logical Clock timestamp of the data update.
+  late final hlcTimestamp = text().map(hlcConverter)();
+
+  /// Raw value of the data.
+  late final rawValue = sqliteAny().nullable()();
+
+  @override
+  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/database/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, #lastReceivedHlc},
+)
+@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};
+}

lib/src/database/triggers.dart

@@ -0,0 +1,148 @@
+import 'package:drift/drift.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].
+  OfflineSyncTriggers(super.crdtDb);
+
+  /// Wraps the provided SQL statement as a CREATE TRIGGER statement.
+  ///
+  /// The received [innerSql] will be expected to go inside a BEGIN...END block.
+  String wrapAsCreateTriggerStatement({
+    required Operation operation,
+    required String tableName,
+    required String triggerName,
+    required String innerSql,
+  });
+
+  /// 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}';
+
+  /// Returns the SQL statement to insert a new row into the CRDT data table.
+  ///
+  /// This statement is compatible with all major SQL dialects like SQLite,
+  /// Postgres, MySQL, SQL Server, etc. So the dialect-specific subclass only
+  /// needs to wrap it in the appropriate CREATE TRIGGER statement.
+  ///
+  /// For some dialects, it will only be needed to enable support for quoted
+  /// identifiers that are used in the statement for maximum compatibility. On
+  /// MySQL this is done by setting the `SQL_MODE` to `ANSI_QUOTES` and on SQL
+  /// Server this is done by setting the `QUOTED_IDENTIFIER` option to `ON`.
+  String _getInsertStatement({
+    required TableInfo<Table, Object?> table,
+    required Operation operation,
+  }) {
+    final tableName = table.actualTableName;
+
+    final uniqueRowId = getUniqueRowId(
+      table,
+      tableAlias: operation == Operation.insert ? 'NEW' : 'OLD',
+    );
+
+    final columnNamesWithoutPrimaryKey = operation == Operation.delete
+        ? [isDeletedColumnName]
+        : table.$columns
+              .map((c) => c.$name)
+              // MAYBE: If we remove the primary key columns from the list, some tables
+              // might be completely ignored (like many-to-many relationship tables).
+              // Do we really need them? Probably yes if we want to avoid deserializing
+              // the split rowId into the original primary key columns, which can have
+              // different types.
+              // .where((c) => !table.$primaryKey.map((p) => p.$name).contains(c))
+              .toList();
+
+    final columnInserts = columnNamesWithoutPrimaryKey.map((c) {
+      return '    SELECT\n${[
+        '      \'$c\' AS "column_name"',
+        '      ${c == isDeletedColumnName ? 'TRUE' : 'NEW."$c"'} AS "raw_value"',
+        if (operation == Operation.update) '      NEW."$c" IS NOT OLD."$c" AS "value_changed"',
+      ].join(',\n')}';
+    });
+
+    final whereClause = operation == Operation.update
+        ? '\n  WHERE (value_changed = TRUE)'
+        : '';
+
+    return '''
+  INSERT OR REPLACE INTO $crdtDataTableName (
+    ${crdtDataAllColumns.join(', ')}
+  )
+  SELECT
+    '${crdtDb.userId}' AS "user_id",
+    '$tableName' AS "table_name",
+    column_name,
+    $uniqueRowId AS "row_id",
+    "hlc_timestamp",
+    raw_value
+  FROM (
+    SELECT $nextHlcFunction('${crdtDb.userId}', '${crdtDb.nodeId}') AS "hlc_timestamp"
+  ), (
+${columnInserts.join('\n    UNION ALL\n')}
+  )$whereClause;''';
+  }
+
+  /// Returns the SQL statements to create the triggers for the changelog table.
+  List<String> generateCreateTriggerStatements(TableInfo<Table, Object?> table) {
+    return [
+      for (final op in Operation.values)
+        wrapAsCreateTriggerStatement(
+          operation: op,
+          tableName: table.actualTableName,
+          triggerName: _getTriggerName(table, op),
+          innerSql: _getInsertStatement(
+            table: table,
+            operation: op,
+          ),
+        ),
+    ];
+  }
+}
+
+/// Provides the CRDT control trigger statements for the SQLite3 database.
+class Sqlite3OfflineSyncTriggers extends OfflineSyncTriggers {
+  /// Creates a new instance of [Sqlite3OfflineSyncTriggers].
+  Sqlite3OfflineSyncTriggers(super.crdtDb);
+
+  @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"
+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.
+  Future<List<String>> getCrdtTriggers() async {
+    final result = await customSelect(
+      '''\
+SELECT type, name, sql
+FROM sqlite_master
+WHERE (type = "trigger")
+  AND (name LIKE "%%__crdt__%%");''',
+    ).get();
+
+    return result.map((row) => row.read<String>('name')).toList();
+  }
+}

lib/src/migrator.dart

@@ -1,8 +1,8 @@
 import 'package:drift/drift.dart';
 
-import '/src/database.dart';
+import 'crdt/crdt.dart';
-import 'crdt.dart';
+import 'database/database.dart';
-import 'triggers.dart';
+import 'database/triggers.dart';
 
 /// Runs migrations declared by a [MigrationStrategy].
 class OfflineSyncMigrator extends Migrator {

lib/src/utils/sql_builder.dart

@@ -1,6 +1,6 @@
 import 'package:drift/drift.dart';
 
-import '/src/database.dart';
+import '../database/database.dart';
 
 /// Provides the base building blocks for SQL statements to interact with the
 /// CRDT data table.

test/database_setup_test.dart

@@ -1,7 +1,7 @@
 import 'package:drift/drift.dart';
 import 'package:drift/native.dart';
 import 'package:drift_offline_first/drift_offline_first.dart';
-import 'package:drift_offline_first/src/triggers.dart';
+import 'package:drift_offline_first/src/database/triggers.dart';
 import 'package:test/test.dart';
 
 void main() {

test/triggers/trigger_creation_test.dart

@@ -1,4 +1,4 @@
-import 'package:drift_offline_first/src/triggers.dart';
+import 'package:drift_offline_first/src/database/triggers.dart';
 import 'package:test/test.dart';
 
 import '../utils/database.dart';

test/triggers/trigger_generation_test.dart

@@ -1,5 +1,5 @@
+import 'package:drift_offline_first/src/database/triggers.dart';
 import 'package:drift_offline_first/src/migrator.dart';
-import 'package:drift_offline_first/src/triggers.dart';
 import 'package:test/test.dart';
 
 import '../utils/database.dart';