feat: First version with working triggers to track tables

.gitignore

@@ -0,0 +1,10 @@
+# https://dart.dev/guides/libraries/private-files
+# Created by `dart pub`
+.dart_tool/
+
+# Avoid committing pubspec.lock for library packages; see
+# https://dart.dev/guides/libraries/private-files#pubspeclock.
+pubspec.lock
+
+.vscode/
+*.g.dart

CHANGELOG.md

@@ -0,0 +1,3 @@
+## 0.0.1
+
+- Initial version.

README.md

@@ -0,0 +1,48 @@
+<!--
+This README describes the package. If you publish this package to pub.dev,
+this README's contents appear on the landing page for your package.
+
+For information about how to write a good package README, see the guide for
+[writing package pages](https://dart.dev/tools/pub/writing-package-pages).
+
+For general information about developing packages, see the Dart guide for
+[creating packages](https://dart.dev/guides/libraries/create-packages)
+and the Flutter guide for
+[developing packages and plugins](https://flutter.dev/to/develop-packages).
+-->
+
+A plugin to build offline-first applications with Drift. Develop your app as a
+normal offline app and use this non-invasive plugin to automatically handle the
+sync process using CRDT (Conflict-free Replicated Data Type). Works
+out-of-the-box with Drift's existing APIs.
+
+## Features
+
+TODO: List what your package can do. Maybe include GIFs, or videos.
+
+## Getting started
+
+TODO: List prerequisites and provide or point to information on how to start
+using the package.
+
+## Usage
+
+TODO: Include short and useful examples for package users. Add longer examples
+to `/example` folder.
+
+See the `/example` directory for a...
+
+```dart
+const like = 'sample';
+```
+
+<!-- ## Additional information
+
+TODO: Tell users more about the package: where to find more information, how to
+contribute to the package, how to file issues, what response they can expect
+from the package authors, and more. -->
+
+## License
+
+This package is distributed under the MIT License. See the LICENSE file for
+more information.

SPEC.md

@@ -0,0 +1,41 @@
+# Specification for the implementation
+
+## Starting the database
+
+- After the `createAll()`, call a custom function to inspect the database and
+  list:
+  - Each table unique constraints
+  - All foreign keys
+
+Check
+[this research](https://www.perplexity.ai/search/on-drift-for-flutter-after-all-aaj9LV1QTeSST3brpolEMA)
+for reference.
+
+## How to specify which constraint to apply to each case?
+
+Unique can be handled as:
+
+- on_conflict_merge (default)
+- on_conflict_overwrite
+
+Foreign keys can be handled as:
+
+- on_fk_deleted_delete_this (default, if cascade delete)
+- on_fk_deleted_set_null (default, if optional)
+- on_fk_deleted_restore_related
+
+## After merge on sync, a callback must be available to check custom rules
+
+There must have an easy way to issue fixes and re-run the merge.
+
+## Next:
+
+- [ ] Design the registration of constraints in the database.
+- [ ] Integrate the CRDT for retrieval of changes.
+- [ ] Integrate the CRDT merge logic.
+- [ ] Integrate the compensation logic with the merge logic.
+- [ ] Implement a regenerate process to run after migrations.
+- [ ] Add pending tests marked as TODO.
+- [ ] Transform this repo in a monorepo with this package and a Flutter version
+      that provides a custom `driftDatabase` function that already passes the
+      closure to add the `next_hlc_timestamp` function to the database.

analysis_options.yaml

@@ -0,0 +1,133 @@
+include: package:lints/recommended.yaml
+
+linter:
+  rules:
+    - always_declare_return_types
+    - always_put_required_named_parameters_first
+    - annotate_redeclares
+    - avoid_bool_literals_in_conditional_expressions
+    - avoid_catches_without_on_clauses
+    - avoid_catching_errors
+    - avoid_empty_else
+    - avoid_escaping_inner_quotes
+    - avoid_field_initializers_in_const_classes
+    - avoid_final_parameters
+    - avoid_js_rounded_ints
+    - avoid_positional_boolean_parameters
+    - avoid_private_typedef_functions
+    - avoid_relative_lib_imports
+    - avoid_returning_this
+    - avoid_setters_without_getters
+    - avoid_shadowing_type_parameters
+    - avoid_slow_async_io
+    - avoid_types_as_parameter_names
+    - avoid_unused_constructor_parameters
+    - avoid_void_async
+    - await_only_futures
+    - camel_case_extensions
+    - cancel_subscriptions
+    - cascade_invocations
+    - cast_nullable_to_non_nullable
+    - close_sinks
+    - collection_methods_unrelated_type
+    - combinators_ordering
+    - comment_references
+    - conditional_uri_does_not_exist
+    - curly_braces_in_flow_control_structures
+    - dangling_library_doc_comments
+    - depend_on_referenced_packages
+    - deprecated_consistency
+    - deprecated_member_use_from_same_package
+    - directives_ordering
+    - discarded_futures
+    - document_ignores
+    - empty_catches
+    - eol_at_end_of_file
+    - file_names
+    - hash_and_equals
+    - implicit_call_tearoffs
+    - invalid_case_patterns
+    - library_names
+    - literal_only_boolean_expressions
+    - matching_super_parameters
+    - missing_code_block_language_in_doc_comment
+    - missing_whitespace_between_adjacent_strings
+    - no_adjacent_strings_in_list
+    - no_duplicate_case_values
+    - no_literal_bool_comparisons
+    - no_runtimeType_toString
+    - no_self_assignments
+    - no_wildcard_variable_uses
+    - non_constant_identifier_names
+    - noop_primitive_operations
+    - null_check_on_nullable_type_parameter
+    - omit_local_variable_types
+    - one_member_abstracts
+    - only_throw_errors
+    - package_prefixed_library_names
+    - parameter_assignments
+    - prefer_const_constructors
+    - prefer_const_declarations
+    - prefer_final_in_for_each
+    - prefer_final_locals
+    - prefer_foreach
+    - prefer_if_elements_to_conditional_expressions
+    - prefer_is_empty
+    - prefer_is_not_empty
+    - prefer_iterable_whereType
+    - prefer_mixin
+    - prefer_null_aware_method_calls
+    - prefer_relative_imports
+    - prefer_single_quotes
+    - prefer_typing_uninitialized_variables
+    - prefer_void_to_null
+    - provide_deprecation_message
+    - public_member_api_docs
+    - require_trailing_commas
+    - sized_box_shrink_expand
+    - sort_constructors_first
+    - sort_unnamed_constructors_first
+    - test_types_in_equals
+    - throw_in_finally
+    - tighten_type_of_initializing_formals
+    - type_annotate_public_apis
+    - type_literal_in_constant_pattern
+    - unawaited_futures
+    - unintended_html_in_doc_comment
+    - unnecessary_await_in_return
+    - unnecessary_breaks
+    - unnecessary_lambdas
+    - unnecessary_library_directive
+    - unnecessary_null_aware_operator_on_extension_on_nullable
+    - unnecessary_null_checks
+    - unnecessary_overrides
+    - unnecessary_parenthesis
+    - unnecessary_raw_strings
+    - unnecessary_statements
+    - unreachable_from_main
+    - unrelated_type_equality_checks
+    - use_colored_box
+    - use_decorated_box
+    - use_enums
+    - use_if_null_to_convert_nulls_to_bools
+    - use_is_even_rather_than_modulo
+    - use_named_constants
+    - use_raw_strings
+    - use_setters_to_change_properties
+    - use_string_buffers
+    - use_string_in_part_of_directives
+    - use_test_throws_matchers
+    - use_to_and_as_if_applicable
+    - valid_regexps
+    - void_checks
+
+formatter:
+  page_width: 88
+
+analyzer:
+  exclude:
+    - "**/*.g.dart"
+    - "**/*.drift.dart"
+  language:
+    strict-casts: true
+    strict-inference: true

build.yaml

@@ -0,0 +1,10 @@
+targets:
+  $default:
+    builders:
+      drift_dev:
+        options:
+          use_sql_column_name_as_json_key: true
+          sql:
+            dialects:
+              - sqlite
+              - postgres

example/drift_offline_first_example.dart

@@ -0,0 +1,5 @@
+import 'package:drift_offline_first/drift_offline_first.dart';
+
+void main() {
+  print('awesome: $OfflineSyncMigrator');
+}

lib/drift_offline_first.dart

@@ -0,0 +1,9 @@
+/// Support for doing something awesome.
+///
+/// More docs go here.
+library;
+
+export 'src/database.dart';
+export 'src/hlc/converter.dart';
+export 'src/migrator.dart';
+export 'src/setup.dart';

lib/src/database.dart

@@ -0,0 +1,29 @@
+import 'package:drift/drift.dart';
+
+import 'tables/compensation.dart';
+import 'tables/control.dart';
+import 'tables/data.dart';
+
+part 'database.g.dart';
+
+/// The main database class for CRDT synchronization.
+@DriftDatabase(
+  tables: [
+    CrdtDataTable,
+    CrdtControlTable,
+    CrdtCompensationTable,
+  ],
+)
+class CrdtDatabase extends _$CrdtDatabase {
+  /// Creates a new instance of [CrdtDatabase] with the provided [executor].
+  CrdtDatabase(super.e);
+
+  /// 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;
+}

lib/src/hlc/converter.dart

@@ -0,0 +1,20 @@
+import 'package:crdt/crdt.dart';
+import 'package:drift/drift.dart';
+
+export 'package:crdt/crdt.dart' show Hlc;
+
+class _HlcConverter extends TypeConverter<Hlc, String> {
+  const _HlcConverter();
+
+  @override
+  Hlc fromSql(String fromDb) => Hlc.parse(fromDb);
+
+  @override
+  String toSql(Hlc value) => value.toJson();
+}
+
+/// Converter for [Hlc] to store it in the database as string.
+const hlcConverter = _HlcConverter();
+
+/// Null-aware converter for [Hlc] to handle nullable fields.
+const nullableHlcConverter = NullAwareTypeConverter.wrap(hlcConverter);

lib/src/hlc/stateful.dart

@@ -0,0 +1,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);
+
+  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));
+  }
+
+  /// 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 nextHlc() {
+    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

@@ -0,0 +1,244 @@
+import 'package:drift/drift.dart';
+
+import '/src/database.dart';
+import 'triggers.dart';
+
+/// Runs migrations declared by a [MigrationStrategy].
+class OfflineSyncMigrator extends Migrator {
+  /// Used internally by drift when opening the database.
+  OfflineSyncMigrator(
+    super.database, {
+    required this.userId,
+    required this.nodeId,
+
+    /// Restrict tables to synchronize. If not provided, all tables will be considered.
+    List<TableInfo>? synchronizedTables,
+
+    /// Optionally specify tables to exclude from synchronization.
+    List<TableInfo>? excludeTables,
+  }) : synchronizedTables = synchronizedTables ?? database.allTables.toList() {
+    if (excludeTables != null) {
+      this.synchronizedTables.removeWhere(excludeTables.contains);
+    }
+    _validateSynchronizedTables();
+  }
+
+  /// The user ID for the CRDT system.
+  ///
+  /// Note that this is the user ID, not the node ID. A user can have multiple nodes,
+  /// all sharing the same user ID so that they can synchronize with each other.
+  final String userId;
+
+  /// The node ID for the CRDT system.
+  ///
+  /// A node is a unique identifier for a device or application instance.
+  final String nodeId;
+
+  /// Tables to be synchronized with CRDT.
+  final List<TableInfo> synchronizedTables;
+
+  /// The database instance to apply the CRDT schema to.
+  late final crdtDb = CrdtDatabase(database.executor);
+
+  /// Migrations for tables that are part of the CRDT system.
+  final pendingMigrations = <String, CRDTTableMigration>{};
+
+  /// Eagerly validates that all synchronized tables are supported.
+  ///
+  /// This prevents errors at runtime later if tables do not meet the requirements.
+  void _validateSynchronizedTables() {
+    for (final table in synchronizedTables) {
+      if (table.$primaryKey.isEmpty) {
+        throw ArgumentError(
+          'Table "${table.actualTableName}" has no primary key and is therefore not '
+          'supported to track changes to it. Either add a primary key to the table '
+          'or remove it from the synchronized tables list.',
+        );
+      }
+    }
+  }
+
+  /// Gets the migration for a table that is part of the CRDT system.
+  ///
+  /// If the table is not in the list of synchronized tables, a new migration
+  /// will be created.
+  CRDTTableMigration _register(String tableName) {
+    return pendingMigrations.putIfAbsent(
+      tableName,
+      () => CRDTTableMigration(tableName),
+    );
+  }
+
+  /// Creates the CRDT-related entities in the database.
+  Future<void> createCrdtControl() async {
+    for (final table in crdtDb.allTables) {
+      await createTable(table);
+    }
+
+    if (crdtDb.isPostgres) {
+      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});',
+      );
+    }
+
+    for (final table in synchronizedTables) {
+      await _createTriggers(table);
+    }
+  }
+
+  Future<void> _createTriggers(TableInfo table) async {
+    final triggerCreator = switch (database.executor.dialect) {
+      SqlDialect.sqlite => Sqlite3OfflineSyncTriggers(this),
+      _ => throw UnsupportedError('Unsupported dialect: ${database.executor.dialect}'),
+    };
+    final triggers = triggerCreator.generateCreateTriggerStatements(table);
+    for (final trigger in triggers) {
+      await database.customStatement(trigger);
+    }
+  }
+
+  @override
+  Future<void> createAll() async {
+    await super.createAll();
+    await createCrdtControl();
+  }
+
+  @override
+  Future<void> createTable(TableInfo table) async {
+    // Tables that start with 'tmp_for_copy_' are used in alter table operations.
+    // We do not want to register them as created tables.
+    if (!table.actualTableName.startsWith('tmp_for_copy_')) {
+      _register(table.actualTableName).created = true;
+    }
+    await super.createTable(table);
+    if (synchronizedTables.contains(table)) {
+      await _createTriggers(table);
+    }
+  }
+
+  @override
+  Future<void> alterTable(TableMigration migration) async {
+    final table = migration.affectedTable;
+    pendingMigrations.putIfAbsent(
+      table.actualTableName,
+      () => CRDTTableMigration.from(migration),
+    );
+
+    final existingTriggers = await database.getCrdtTriggers();
+    for (final trigger in existingTriggers) {
+      await database.customStatement('DROP TRIGGER IF EXISTS $trigger;');
+    }
+
+    await super.alterTable(migration);
+    await _createTriggers(table);
+  }
+
+  @override
+  Future<void> drop(DatabaseSchemaEntity entity) async {
+    if (entity is TableInfo) {
+      _register(entity.actualTableName).dropped = true;
+    }
+    return super.drop(entity);
+  }
+
+  @override
+  Future<void> deleteTable(String name) async {
+    _register(name).dropped = true;
+    return super.deleteTable(name);
+  }
+
+  @override
+  Future<void> addColumn(TableInfo table, GeneratedColumn column) async {
+    _register(table.actualTableName).onColumn(column.$name).added = true;
+    return super.addColumn(table, column);
+  }
+
+  @override
+  Future<void> dropColumn(TableInfo table, String column) async {
+    _register(table.actualTableName).onColumn(column).dropped = true;
+    return super.dropColumn(table, column);
+  }
+
+  @override
+  Future<void> renameColumn(
+    TableInfo table,
+    String oldName,
+    GeneratedColumn column,
+  ) async {
+    _register(table.actualTableName).onColumn(column.$name).oldName = oldName;
+    return super.renameColumn(table, oldName, column);
+  }
+
+  @override
+  Future<void> renameTable(TableInfo table, String oldName) async {
+    _register(table.actualTableName).oldName = oldName;
+    return super.renameTable(table, oldName);
+  }
+}
+
+/// A migration for a table that is part of the CRDT system.
+class CRDTTableMigration {
+  /// Creates a new instance of [CRDTTableMigration].
+  CRDTTableMigration(this.tableName);
+
+  /// The table that is being migrated.
+  final String tableName;
+
+  /// Whether the table was created.
+  bool created = false;
+
+  /// Whether the table was dropped.
+  bool dropped = false;
+
+  /// The old name of the table, if the table was renamed.
+  String? oldName;
+
+  /// The columns that were added.
+  final columns = <String, CRDTColumnMigration>{};
+
+  /// Whether the table was modified.
+  bool get modified => oldName != null || columns.isNotEmpty;
+
+  /// Gets the migration for a column that is part of the CRDT system.
+  ///
+  /// If the column is not in the list of synchronized columns, a new migration
+  /// will be created.
+  CRDTColumnMigration onColumn(String columnName) {
+    return columns.putIfAbsent(columnName, () => CRDTColumnMigration(columnName));
+  }
+
+  /// Creates a new instance of [CRDTTableMigration] from a [TableMigration].
+  static CRDTTableMigration from(TableMigration migration) {
+    final table = migration.affectedTable;
+    final tableMigration = CRDTTableMigration(table.actualTableName);
+    for (final column in migration.newColumns) {
+      // TODO: Handle the column transformer.
+      tableMigration.onColumn(column.name);
+    }
+    return tableMigration;
+  }
+}
+
+/// A migration for a table that is part of the CRDT system.
+class CRDTColumnMigration {
+  /// Creates a new instance of [CRDTColumnMigration].
+  CRDTColumnMigration(this.columnName);
+
+  /// The column that is being migrated.
+  final String columnName;
+
+  /// Whether the column was added.
+  bool added = false;
+
+  /// Whether the column was dropped.
+  bool dropped = false;
+
+  /// The old name of the column, if the column was renamed.
+  String? oldName;
+
+  /// Whether the column type was modified.
+  bool changedType = false;
+}

lib/src/setup.dart

@@ -0,0 +1,32 @@
+import 'package:drift/native.dart';
+
+import '/src/hlc/stateful.dart';
+import '/src/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 {
+  // if (_initialized) return (database) {};
+  // _initialized = true;
+
+  return (database) {
+    database.createFunction(
+      functionName: nextHlcFunction,
+      function: (args) {
+        final nodeId = args.firstOrNull as String?;
+        if (nodeId == null) throw ArgumentError('nodeId is required');
+        return StatefulHlc.cached(nodeId).nextHlc().toString();
+      },
+      // argumentCount: const AllowedArgumentCount(1),
+      deterministic: false,
+      directOnly: false,
+    );
+  };
+}

lib/src/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/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/tables/data.dart

@@ -0,0 +1,58 @@
+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.
+///
+@DataClassName('CrdtDataEntry')
+class CrdtDataTable extends Table {
+  @override
+  String get tableName => '__crdt_data';
+
+  @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()();
+
+  /// 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()();
+
+  /// Boolean indicating if the value has changed since last sync.
+  late final valueChanged = boolean().withDefault(const Constant(true))();
+
+  @override
+  Set<Column> get primaryKey => {/*userId,*/ tblName, columnName, rowId};
+}

lib/src/triggers.dart

@@ -0,0 +1,189 @@
+import 'package:drift/drift.dart';
+
+import '/src/migrator.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 {
+  /// Creates a new instance of [OfflineSyncTriggers] with the provided [migrator].
+  const OfflineSyncTriggers(this.migrator);
+
+  /// 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.
+  ///
+  /// 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 =
+        table.getUniqueRowId(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) {
+      final rawValue = c == isDeletedColumnName ? 'TRUE' : 'NEW."$c"';
+      final valueChanged =
+          operation != Operation.update ? 'TRUE' : 'NEW."$c" IS NOT OLD."$c"';
+
+      return '''
+    SELECT
+      '$c' AS "column_name",
+      $rawValue AS "raw_value",
+      $valueChanged AS "value_changed"''';
+    });
+
+    final whereClause =
+        operation == Operation.update ? '\n  WHERE (value_changed = TRUE)' : '';
+
+    return '''
+  INSERT OR REPLACE INTO $_crdtDataTableName (
+    ${_crdtDataAllColumns.join(', ')}
+  )
+  SELECT
+    '$tableName' AS "table_name",
+    column_name,
+    $uniqueRowId AS "row_id",
+    "hlc_timestamp",
+    raw_value,
+    value_changed
+  FROM (
+    SELECT $nextHlcFunction('${migrator.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.migrator);
+
+  @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 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.
+  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();
+  }
+}

pubspec.yaml

@@ -0,0 +1,19 @@
+name: drift_offline_first
+description: A Drift plugin to build offline-first applications with CRDT support.
+version: 0.0.1
+repository: https://github.com/marcelomendoncasoares/drift_offline_first
+
+environment:
+  sdk: ^3.3.0
+
+# TODO: Relax the package constraints to expand compatibility.
+dependencies:
+  crdt: ^5.1.3
+  drift: ^2.28.0
+  uuid: ^4.5.1
+
+dev_dependencies:
+  build_runner: ^2.6.0
+  drift_dev: ^2.28.0
+  lints: ^6.0.0
+  test: ^1.26.2

query_examples.sql

@@ -0,0 +1,98 @@
+UPDATE crdt_metadata
+SET `value` = d.`value`
+FROM crdt_metadata AS s
+JOIN (
+  -- Set multiple fields of the entities that match the where clause
+  SELECT '__is_deleted__' AS `column`, 'TRUE' AS `value`
+  UNION ALL
+  SELECT 'reference_id', NULL;
+) AS d ON s.`column` = d.`column`
+WHERE (
+  `table` = 'category' AND
+  `column` = 'parent_id' AND
+  `value` = 'THIS_ENTITY_ROW_ID'
+);
+
+
+INSERT INTO crdt_metadata ("table", "column", "value")
+SELECT
+  'category'            AS "table",
+  d."column"            AS "column",
+  d."value"             AS "value"
+FROM (
+  SELECT '__is_deleted__' AS "column", 'TRUE' AS "value"
+  UNION ALL
+  SELECT 'reference_id', NULL
+) AS d
+ON CONFLICT("table", "column") DO UPDATE
+SET "value" = EXCLUDED."value";
+
+INSERT INTO crdt_metadata ("table", "column", "value")
+VALUES
+  ('category', '__is_deleted__', 'TRUE'),
+  ('category', 'reference_id', NULL)
+ON CONFLICT("table", "column") DO UPDATE
+SET "value" = excluded."value";
+
+
+INSERT INTO __crdt_data (
+  'user_id',
+  'table_name',
+  'column_name',
+  'row_id',
+  'hlc_timestamp',
+  'raw_value',
+  'value_changed'
+)
+VALUES
+  (?, 'category', '__is_deleted__', ?, ?, ?, 1),
+  (?, 'category', 'reference_id', ?, ?, ?, 1)
+ON CONFLICT ('user_id', 'table_name', 'column_name', 'row_id') DO UPDATE
+SET
+  'hlc_timestamp' = excluded.'hlc_timestamp',
+  'raw_value'     = excluded.'raw_value',
+  'value_changed' = excluded.'value_changed';
+
+
+
+WITH hlc AS (SELECT $nextHlcFunction() AS ts)
+INSERT INTO __crdt_data (
+  'table_name', 'column_name', 'row_id', 'hlc_timestamp', 'raw_value', 'value_changed'
+)
+SELECT
+  '$table_name', d.'column_name', OLD.'$id_field', hlc.ts, d.'raw_value', d.'value_changed'
+FROM (
+  SELECT
+    '$column_name' AS 'column_name',
+    '?1' AS 'row_id',
+    '?2' AS 'raw_value',
+    'TRUE' AS 'value_changed'
+) as d, hlc
+ON CONFLICT ('table_name', 'column_name', 'row_id') DO UPDATE
+SET
+  'hlc_timestamp' = excluded.'hlc_timestamp',
+  'raw_value'     = excluded.'raw_value',
+  'value_changed' = excluded.'value_changed';
+
+
+CREATE TRIGGER my_trigger
+AFTER INSERT ON '$table_name'
+BEGIN
+  INSERT INTO __crdt_data (
+    'table_name', 'column_name', 'row_id', 'hlc_timestamp', 'raw_value', 'value_changed'
+  )
+  SELECT * FROM (
+    WITH hlc AS (SELECT $nextHlcFunction() AS ts)
+    SELECT
+      '$table_name' AS 'table_name',
+      '$column_name' AS 'column_name',
+      OLD.'$id_field' AS 'row_id',
+      hlc.ts AS 'hlc_timestamp',
+      NEW.'$column_name' AS 'raw_value',
+      NEW.'$column_name' IS NOT OLD.'$column_name' AS 'value_changed'
+    FROM hlc
+    UNION ALL
+    [repeat for each column]
+  )
+  WHERE NEW.'$column_name' IS NOT OLD.'$column_name';
+END;

test/triggers/trigger_creation_test.dart

@@ -0,0 +1,35 @@
+import 'package:drift_offline_first/src/triggers.dart';
+import 'package:test/test.dart';
+
+import '../utils/database.dart';
+
+void main() {
+  group('Given a database with an OfflineSyncMigrator', () {
+    group('when running the migration', () {
+      setUp(() async {
+        // Insert a new row to trigger the migration.
+        await database.managers.todosTable.create(
+          (t) => t(content: 'test'),
+        );
+      });
+
+      test('then the triggers are created for every synchronized table.', () async {
+        final allTriggerNames = await database.getCrdtTriggers();
+
+        for (final table in database.createMigrator().synchronizedTables) {
+          final triggerNames = allTriggerNames
+              .where((name) => name.startsWith('__crdt__${table.actualTableName}__'))
+              .toList();
+
+          expect(triggerNames, contains('__crdt__${table.actualTableName}__insert'));
+          expect(triggerNames, contains('__crdt__${table.actualTableName}__update'));
+          expect(triggerNames, contains('__crdt__${table.actualTableName}__delete'));
+        }
+      });
+    });
+  });
+
+  // TODO: Add a test for the case where a table has no primary key.
+  // TODO: Add a test for the case where a table has multiple primary key columns.
+  // TODO: Add a test for the case where a table has only primary key columns.
+}