feat: Implement merge changes method (#38)

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: Marcelo Soares <marcelo.me.soares@gmail.com>

packages/serverpod_offline_sync_client/lib/crdt.dart

@@ -1,3 +1,4 @@
+export 'src/crdt/merge.dart';
 export 'src/database/database.dart';
 export 'src/database/recorder.dart';
 export 'src/database/schema.dart';

packages/serverpod_offline_sync_client/lib/serverpod_offline_sync_client.dart

@@ -1,4 +1,5 @@
 export 'package:serverpod_client/serverpod_client.dart';
 
 export 'crdt.dart';
+export 'src/crdt/extensions.dart';
 export 'src/protocol/protocol.dart';

packages/serverpod_offline_sync_client/lib/src/crdt/extensions.dart

@@ -0,0 +1,33 @@
+import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart';
+
+import '../protocol/protocol.dart';
+
+/// Extensions for the [CrdtDataField] class.
+extension CrdtDataFieldExtension on CrdtDataField {
+  /// The HLC represented by this field.
+  Hlc get hlc => node == null
+      ? (throw StateError(
+          'Fetch the CRDT field with the node included to get the HLC.',
+        ))
+      : toHlcForNode(node!.uuidNodeId);
+}
+
+/// Extensions for the [CrdtDataRow] class.
+extension CrdtDataRowExtension on CrdtDataRow {
+  /// The HLC represented by this row.
+  Hlc get hlc => node == null
+      ? (throw StateError(
+          'Fetch the CRDT row with the node included to get the HLC.',
+        ))
+      : toHlcForNode(node!.uuidNodeId);
+}
+
+/// Extensions for the [CrdtDataDeleted] class.
+extension CrdtDataDeletedExtension on CrdtDataDeleted {
+  /// The HLC represented by this tombstone.
+  Hlc get hlc => node == null
+      ? (throw StateError(
+          'Fetch the CRDT tombstone with the node included to get the HLC.',
+        ))
+      : toHlcForNode(node!.uuidNodeId);
+}

packages/serverpod_offline_sync_client/lib/src/crdt/merge.dart

@@ -0,0 +1,81 @@
+import 'package:serverpod_client/serverpod_client.dart';
+import 'package:serverpod_database/serverpod_database.dart';
+import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart';
+
+import '../protocol/protocol.dart';
+
+/// Merge metadata extracted from a [CrdtMergeSet].
+typedef CrdtMergeMetadataLookup = ({
+  Map<String, Set<UuidValue>> rowIdsByTable,
+  Map<String, Set<String>> columnNamesByTable,
+});
+
+/// CRDT merge helpers built on top of the generated Serverpod models.
+extension CrdtMergeSetExtension on CrdtMergeSet {
+  /// Whether this merge set has no changes.
+  bool get isEmpty => inserts.isEmpty && updates.isEmpty && deletes.isEmpty;
+
+  /// All merge changes in this set.
+  Iterable<CrdtMergeChange> get changes sync* {
+    yield* inserts;
+    yield* updates;
+    yield* deletes;
+  }
+
+  /// All merge changes in this set sorted by causal order.
+  List<CrdtMergeChange> get causallyOrderedChanges {
+    final operations = changes.toList()
+      ..sort((left, right) => left.hlc.compareTo(right.hlc));
+    return operations;
+  }
+
+  /// Collects the row and column metadata needed to load merge state.
+  CrdtMergeMetadataLookup collectMetadataLookup({
+    required Map<String, Set<String>> columnNamesByTableName,
+  }) {
+    final rowIdsToLoadByTable = <String, Set<UuidValue>>{};
+    final columnNamesToLoadByTable = <String, Set<String>>{};
+
+    for (final change in changes) {
+      final availableColumnNames = columnNamesByTableName[change.tableName];
+      if (availableColumnNames == null) continue;
+
+      rowIdsToLoadByTable.putIfAbsent(change.tableName, () => {}).add(change.uuidRowId);
+      columnNamesToLoadByTable.putIfAbsent(change.tableName, () => {}).addAll(
+        switch (change) {
+          CrdtMergeInsert() => availableColumnNames.difference({'id'}),
+          CrdtMergeUpdate() => [change.columnName],
+          _ => [],
+        },
+      );
+    }
+
+    return (
+      rowIdsByTable: rowIdsToLoadByTable,
+      columnNamesByTable: columnNamesToLoadByTable,
+    );
+  }
+}
+
+/// Convenience helpers shared by every merge change.
+extension CrdtMergeChangeExtension on CrdtMergeChange {
+  /// The HLC represented by this change.
+  Hlc get hlc => Hlc(hlcDatetime, hlcCounter, uuidNodeId);
+}
+
+/// Convenience helpers for insert changes.
+extension CrdtMergeInsertExtension on CrdtMergeInsert {
+  /// The database column payload represented by this change.
+  Map<String, Object?> get databaseColumns {
+    final payload = switch (data) {
+      final TableRow row => row.toJsonForDatabase(),
+      final Map<String, dynamic> map => map,
+      _ => throw StateError(
+        'Unsupported merge insert payload type for $tableName: '
+        '${data.runtimeType}. Expected TableRow or Map<String, dynamic>.',
+      ),
+    };
+
+    return Map<String, Object?>.from(payload as Map<String, dynamic>);
+  }
+}

packages/serverpod_offline_sync_client/lib/src/crdt/sync.dart

@@ -1 +0,0 @@
-// TODO: Implement the business layer for the CRDT sync.

packages/serverpod_offline_sync_client/lib/src/database/database.dart

@@ -5,6 +5,7 @@
 import 'package:serverpod_database/serverpod_database.dart';
 import 'package:uuid/uuid.dart';
 
+import '../crdt/merge.dart';
 import '../managers/user.dart';
 import '../protocol/protocol.dart';
 import 'recorder.dart';
@@ -43,6 +44,29 @@ class CrdtDatabase implements Database {
     await _recorder.initialize();
   }
 
+  /// Merges remote CRDT changes into the local database for the given user.
+  ///
+  /// When [userId] is omitted, this uses the recorder's persistent user id.
+  /// The merge locks the current user's CRDT tables and executes atomically
+  /// inside [transactionForUser].
+  Future<void> mergeChanges(
+    CrdtMergeSet mergeSet, {
+    UuidValue? userId,
+  }) async {
+    if (mergeSet.isEmpty) return;
+
+    final effectiveUserId =
+        userId ??
+        _recorder.persistentUserId ??
+        (throw StateError(
+          'A user ID is required when merging changes without a persistent user.',
+        ));
+    await transactionForUser<void>(effectiveUserId, (tx) async {
+      await _recorder.lockCurrentUser(tx);
+      await _recorder.mergeChanges(mergeSet, tx);
+    });
+  }
+
   @override
   DatabaseAnalyzer get analyzer => _delegate.analyzer;
 

packages/serverpod_offline_sync_client/lib/src/database/merge.dart

@@ -0,0 +1,599 @@
+part of 'recorder.dart';
+
+typedef _MergeRowKey = (String, UuidValue);
+typedef _MergeFieldKey = (String, UuidValue, String);
+
+/// Adds merge-specific behavior to [CrdtMutationRecorder].
+extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
+  /// Locks the current user row so merges can serialize with other work.
+  Future<void> lockCurrentUser(Transaction transaction) async {
+    final user = _getEffectiveUser(transaction);
+    // Use a row lock without fetching the record since the merge path only
+    // needs serialization against concurrent work for the same user.
+    await CrdtUser.db.lockRows(
+      _session,
+      where: (t) => t.id.equals(user.id),
+      transaction: transaction,
+      lockMode: LockMode.forUpdate,
+      lockBehavior: LockBehavior.wait,
+    );
+  }
+
+  /// Merges remote CRDT changes into the current database.
+  Future<void> mergeChanges(
+    CrdtMergeSet mergeSet,
+    Transaction transaction,
+  ) async {
+    if (mergeSet.isEmpty) return;
+
+    final operations = mergeSet.causallyOrderedChanges;
+    final currentUser = _getEffectiveUser(transaction);
+    final remoteNodes = await _findOrCreateNodesForMerge(
+      currentUser.id!,
+      {for (final change in operations) change.uuidNodeId},
+      transaction,
+    );
+    final metadata = await _loadMergeMetadata(mergeSet, transaction);
+
+    for (final operation in operations) {
+      if (!_isCrdtTrackedTableName(operation.tableName)) {
+        continue;
+      }
+
+      switch (operation) {
+        case final CrdtMergeInsert insert:
+          await _applyMergeInsert(
+            insert,
+            remoteNodes,
+            metadata.rows,
+            metadata.fields,
+            metadata.tombstones,
+            transaction,
+          );
+        case final CrdtMergeUpdate update:
+          await _applyMergeUpdate(
+            update,
+            remoteNodes,
+            metadata.rows,
+            metadata.fields,
+            transaction,
+          );
+        case final CrdtMergeDelete delete:
+          await _applyMergeDelete(
+            delete,
+            remoteNodes,
+            metadata.rows,
+            metadata.tombstones,
+            transaction,
+          );
+      }
+    }
+
+    _updateHlcFromIncomingOperations(operations, transaction);
+  }
+
+  void _updateHlcFromIncomingOperations(
+    List<CrdtMergeChange> operations,
+    Transaction transaction,
+  ) {
+    final maxIncomingHlc = operations.fold<Hlc?>(
+      null,
+      (current, change) =>
+          current == null || change.hlc > current ? change.hlc : current,
+    );
+    if (maxIncomingHlc != null) {
+      final hlcManager = _getHlcManager(transaction);
+      if (maxIncomingHlc.nodeId == hlcManager.uuidNodeId) {
+        if (maxIncomingHlc > hlcManager.lastHlc) {
+          hlcManager.lastHlc = maxIncomingHlc;
+        }
+      } else {
+        hlcManager.merge(maxIncomingHlc);
+      }
+    }
+  }
+
+  Future<Map<UuidValue, CrdtNode>> _findOrCreateNodesForMerge(
+    int userId,
+    Set<UuidValue> nodeIds,
+    Transaction transaction,
+  ) async {
+    if (nodeIds.isEmpty) return {};
+
+    var nodes = await CrdtNode.db.find(
+      _session,
+      where: (t) => t.userId.equals(userId) & t.uuidNodeId.inSet(nodeIds),
+      transaction: transaction,
+    );
+    final existingNodeIds = nodes.map((node) => node.uuidNodeId).toSet();
+    final missingNodeIds = nodeIds.difference(existingNodeIds);
+
+    if (missingNodeIds.isNotEmpty) {
+      await CrdtNode.db.insert(
+        _session,
+        [
+          for (final uuidNodeId in missingNodeIds)
+            CrdtNode(
+              userId: userId,
+              uuidNodeId: uuidNodeId,
+            ),
+        ],
+        transaction: transaction,
+        ignoreConflicts: true,
+      );
+
+      nodes = await CrdtNode.db.find(
+        _session,
+        where: (t) => t.userId.equals(userId) & t.uuidNodeId.inSet(nodeIds),
+        transaction: transaction,
+      );
+    }
+
+    return {
+      for (final node in nodes) node.uuidNodeId: node,
+    };
+  }
+
+  Future<
+    ({
+      Map<_MergeRowKey, CrdtDataRow> rows,
+      Map<_MergeFieldKey, CrdtDataField> fields,
+      Map<_MergeRowKey, CrdtDataDeleted> tombstones,
+    })
+  >
+  _loadMergeMetadata(
+    CrdtMergeSet mergeSet,
+    Transaction transaction,
+  ) async {
+    final (:rowIdsByTable, :columnNamesByTable) = mergeSet.collectMetadataLookup(
+      columnNamesByTableName: _syncedTableColumnNamesForMerge,
+    );
+
+    final rows = <_MergeRowKey, CrdtDataRow>{};
+    final fields = <_MergeFieldKey, CrdtDataField>{};
+    final tombstones = <_MergeRowKey, CrdtDataDeleted>{};
+
+    for (final MapEntry(key: tableName, value: rowIds) in rowIdsByTable.entries) {
+      final loadedRows = await _findCrdtRows(
+        tableName,
+        rowIds,
+        transaction,
+        include: CrdtDataRow.include(
+          node: CrdtNode.include(),
+          deleted: CrdtDataDeleted.include(node: CrdtNode.include()),
+        ),
+      );
+
+      for (final row in loadedRows) {
+        rows[(tableName, row.uuidRowId)] = row;
+        if (row.deleted != null) {
+          tombstones[(tableName, row.uuidRowId)] = row.deleted!;
+        }
+      }
+
+      final columnNames = columnNamesByTable[tableName];
+      if (columnNames == null || columnNames.isEmpty || loadedRows.isEmpty) continue;
+
+      final rowPks = loadedRows.map((row) => row.id).whereType<int>().toSet();
+      if (rowPks.length != loadedRows.length) {
+        throw StateError(
+          'Some merge metadata rows for table $tableName are missing '
+          'persisted identifiers.',
+        );
+      }
+
+      final (_, columnsByName) = _schema[tableName]!;
+      final columnIds = <int>{
+        for (final columnName in columnNames)
+          if (columnsByName[columnName] != null) columnsByName[columnName]!.id!,
+      };
+      if (columnIds.isEmpty) continue;
+
+      final loadedFields = await CrdtDataField.db.find(
+        _session,
+        where: (t) => t.rowId.inSet(rowPks) & t.columnId.inSet(columnIds),
+        include: CrdtDataField.include(
+          column: CrdtSchemaColumn.include(),
+          node: CrdtNode.include(),
+        ),
+        transaction: transaction,
+      );
+
+      for (final field in loadedFields) {
+        final row = loadedRows.firstWhere((loadedRow) => loadedRow.id == field.rowId);
+        fields[(tableName, row.uuidRowId, field.column!.name)] = field;
+      }
+    }
+
+    return (
+      rows: rows,
+      fields: fields,
+      tombstones: tombstones,
+    );
+  }
+
+  Future<void> _applyMergeInsert(
+    CrdtMergeInsert insert,
+    Map<UuidValue, CrdtNode> remoteNodes,
+    Map<_MergeRowKey, CrdtDataRow> rows,
+    Map<_MergeFieldKey, CrdtDataField> fields,
+    Map<_MergeRowKey, CrdtDataDeleted> tombstones,
+    Transaction transaction,
+  ) async {
+    final rowKey = (insert.tableName, insert.uuidRowId);
+    final incomingHlc = insert.hlc;
+    final remoteNode = _requireRemoteNode(remoteNodes, insert.uuidNodeId);
+
+    final currentRow = rows[rowKey];
+    if (currentRow != null && incomingHlc <= currentRow.hlc) {
+      return;
+    }
+
+    final data = _sanitizeMergeRowData(insert.tableName, insert.databaseColumns);
+
+    if (currentRow == null) {
+      rows[rowKey] = await _applyMergeInsertForMissingRow(
+        insert,
+        remoteNode,
+        incomingHlc,
+        data,
+        transaction,
+      );
+    } else {
+      await _applyMergeInsertForExistingRow(
+        insert,
+        currentRow,
+        remoteNode,
+        incomingHlc,
+        data,
+        fields,
+        tombstones,
+        transaction,
+      );
+    }
+  }
+
+  Future<void> _applyMergeUpdate(
+    CrdtMergeUpdate update,
+    Map<UuidValue, CrdtNode> remoteNodes,
+    Map<_MergeRowKey, CrdtDataRow> rows,
+    Map<_MergeFieldKey, CrdtDataField> fields,
+    Transaction transaction,
+  ) async {
+    final rowKey = (update.tableName, update.uuidRowId);
+    final row = rows[rowKey];
+    if (row == null) return;
+
+    final shouldApply = await _shouldMergeFieldMetadataIfNewer(
+      tableName: update.tableName,
+      rowId: update.uuidRowId,
+      columnName: update.columnName,
+      row: row,
+      remoteNode: _requireRemoteNode(remoteNodes, update.uuidNodeId),
+      incomingHlc: update.hlc,
+      fields: fields,
+      transaction: transaction,
+    );
+
+    if (!shouldApply) return;
+    await _updateDomainRows(
+      update.tableName,
+      {update.uuidRowId},
+      {update.columnName: update.value},
+      transaction,
+    );
+  }
+
+  Future<void> _applyMergeDelete(
+    CrdtMergeDelete delete,
+    Map<UuidValue, CrdtNode> remoteNodes,
+    Map<_MergeRowKey, CrdtDataRow> rows,
+    Map<_MergeRowKey, CrdtDataDeleted> tombstones,
+    Transaction transaction,
+  ) async {
+    final rowKey = (delete.tableName, delete.uuidRowId);
+    final row = rows[rowKey];
+    if (row == null) return;
+
+    final incomingHlc = delete.hlc;
+    final currentTombstone = tombstones[rowKey];
+    if (incomingHlc <= row.hlc.maxBetween(currentTombstone?.hlc)) {
+      return;
+    }
+
+    tombstones[rowKey] = await _upsertMergeTombstone(
+      row,
+      _requireRemoteNode(remoteNodes, delete.uuidNodeId),
+      incomingHlc,
+      delete.isDeleted,
+      currentTombstone,
+      transaction,
+    );
+  }
+
+  Future<CrdtDataRow> _upsertMergeRow(
+    String tableName,
+    UuidValue rowId,
+    CrdtNode remoteNode,
+    Hlc incomingHlc,
+    CrdtDataRow? currentRow,
+    Transaction transaction,
+  ) async {
+    if (currentRow == null) {
+      final (tableId, _) = _schema[tableName]!;
+      final insertedRow = await CrdtDataRow.db.insertRow(
+        _session,
+        CrdtDataRow(
+          userId: _getHlcManager(transaction).normalizedUserId,
+          tblId: tableId,
+          uuidRowId: rowId,
+          nodeId: remoteNode.id!,
+          hlcDatetime: incomingHlc.datetime,
+          hlcCounter: incomingHlc.counter,
+        ),
+        transaction: transaction,
+      );
+
+      return insertedRow.copyWith(node: remoteNode);
+    }
+
+    final updatedRow = currentRow.copyWith(
+      nodeId: remoteNode.id,
+      node: remoteNode,
+      hlcDatetime: incomingHlc.datetime,
+      hlcCounter: incomingHlc.counter,
+    );
+
+    await CrdtDataRow.db.update(
+      _session,
+      [updatedRow],
+      columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter],
+      transaction: transaction,
+    );
+
+    return updatedRow;
+  }
+
+  Future<CrdtDataField> _upsertMergeField(
+    CrdtDataRow row,
+    CrdtSchemaColumn column,
+    CrdtNode remoteNode,
+    Hlc incomingHlc,
+    CrdtDataField? currentField,
+    Transaction transaction,
+  ) async {
+    if (currentField == null) {
+      final insertedField = await CrdtDataField.db.insertRow(
+        _session,
+        CrdtDataField(
+          rowId: row.id!,
+          columnId: column.id!,
+          nodeId: remoteNode.id!,
+          hlcDatetime: incomingHlc.datetime,
+          hlcCounter: incomingHlc.counter,
+        ),
+        transaction: transaction,
+      );
+
+      return insertedField.copyWith(
+        row: row,
+        column: column,
+        node: remoteNode,
+      );
+    }
+
+    final updatedField = currentField.copyWith(
+      row: row,
+      column: column,
+      nodeId: remoteNode.id,
+      node: remoteNode,
+      hlcDatetime: incomingHlc.datetime,
+      hlcCounter: incomingHlc.counter,
+    );
+
+    await CrdtDataField.db.update(
+      _session,
+      [updatedField],
+      columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter],
+      transaction: transaction,
+    );
+    return updatedField;
+  }
+
+  Future<CrdtDataDeleted> _upsertMergeTombstone(
+    CrdtDataRow row,
+    CrdtNode remoteNode,
+    Hlc incomingHlc,
+    bool isDeleted,
+    CrdtDataDeleted? currentTombstone,
+    Transaction transaction,
+  ) async {
+    if (currentTombstone == null) {
+      final insertedTombstone = await CrdtDataDeleted.db.insertRow(
+        _session,
+        CrdtDataDeleted(
+          rowId: row.id!,
+          nodeId: remoteNode.id!,
+          hlcDatetime: incomingHlc.datetime,
+          hlcCounter: incomingHlc.counter,
+          isDeleted: isDeleted,
+        ),
+        transaction: transaction,
+      );
+
+      return insertedTombstone.copyWith(
+        row: row,
+        node: remoteNode,
+      );
+    }
+
+    final updatedTombstone = currentTombstone.copyWith(
+      row: row,
+      nodeId: remoteNode.id,
+      node: remoteNode,
+      hlcDatetime: incomingHlc.datetime,
+      hlcCounter: incomingHlc.counter,
+      isDeleted: isDeleted,
+    );
+
+    await CrdtDataDeleted.db.update(
+      _session,
+      [updatedTombstone],
+      columns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter, t.isDeleted],
+      transaction: transaction,
+    );
+
+    return updatedTombstone;
+  }
+
+  Future<CrdtDataRow> _applyMergeInsertForMissingRow(
+    CrdtMergeInsert insert,
+    CrdtNode remoteNode,
+    Hlc incomingHlc,
+    Map<String, Object?> data,
+    Transaction transaction,
+  ) async {
+    final insertedCrdtRow = await _upsertMergeRow(
+      insert.tableName,
+      insert.uuidRowId,
+      remoteNode,
+      incomingHlc,
+      null,
+      transaction,
+    );
+
+    final row = _requireMergeInsertTableRow(insert);
+    try {
+      await _db.updateRow(
+        row,
+        columns: _syncTableByName[insert.tableName]!.managedColumns
+            .where((c) => data.containsKey(c.columnName))
+            .toList(),
+        transaction: transaction,
+      );
+    } on DatabaseUpdateRowException {
+      await _db.insertRow(row, transaction: transaction);
+    }
+
+    return insertedCrdtRow;
+  }
+
+  Future<void> _applyMergeInsertForExistingRow(
+    CrdtMergeInsert insert,
+    CrdtDataRow currentRow,
+    CrdtNode remoteNode,
+    Hlc incomingHlc,
+    Map<String, Object?> data,
+    Map<_MergeFieldKey, CrdtDataField> fields,
+    Map<_MergeRowKey, CrdtDataDeleted> tombstones,
+    Transaction transaction,
+  ) async {
+    final (_, columnsByName) = _schema[insert.tableName]!;
+    final updatedValues = <String, Object?>{};
+    for (final MapEntry(key: columnName, value: schemaColumn)
+        in columnsByName.entries) {
+      if (columnName == 'id') continue;
+
+      final shouldApply = await _shouldMergeFieldMetadataIfNewer(
+        tableName: insert.tableName,
+        rowId: insert.uuidRowId,
+        columnName: columnName,
+        row: currentRow,
+        remoteNode: remoteNode,
+        incomingHlc: incomingHlc,
+        fields: fields,
+        transaction: transaction,
+        schemaColumn: schemaColumn,
+      );
+
+      if (!shouldApply) continue;
+      updatedValues[columnName] = data[columnName];
+    }
+
+    if (updatedValues.isNotEmpty) {
+      await _updateDomainRows(
+        insert.tableName,
+        {insert.uuidRowId},
+        updatedValues,
+        transaction,
+      );
+    }
+
+    final rowKey = (insert.tableName, insert.uuidRowId);
+    final currentTombstone = tombstones[rowKey];
+    if (currentTombstone == null || incomingHlc > currentTombstone.hlc) {
+      tombstones[rowKey] = await _upsertMergeTombstone(
+        currentRow,
+        remoteNode,
+        incomingHlc,
+        false,
+        currentTombstone,
+        transaction,
+      );
+    }
+  }
+
+  Future<bool> _shouldMergeFieldMetadataIfNewer({
+    required String tableName,
+    required UuidValue rowId,
+    required String columnName,
+    required CrdtDataRow row,
+    required CrdtNode remoteNode,
+    required Hlc incomingHlc,
+    required Map<_MergeFieldKey, CrdtDataField> fields,
+    required Transaction transaction,
+    CrdtSchemaColumn? schemaColumn,
+  }) async {
+    final (_, columnsByName) = _schema[tableName]!;
+    final resolvedSchemaColumn = schemaColumn ?? columnsByName[columnName];
+    if (resolvedSchemaColumn == null) return false;
+
+    final fieldKey = (tableName, rowId, columnName);
+    final currentField = fields[fieldKey];
+    final currentHlc = currentField?.hlc ?? row.hlc;
+    if (incomingHlc <= currentHlc) return false;
+
+    fields[fieldKey] = await _upsertMergeField(
+      row,
+      resolvedSchemaColumn,
+      remoteNode,
+      incomingHlc,
+      currentField,
+      transaction,
+    );
+    return true;
+  }
+
+  TableRow _requireMergeInsertTableRow(CrdtMergeInsert insert) {
+    final row = insert.data;
+    if (row is TableRow) return row;
+
+    throw StateError(
+      'Unsupported merge insert payload type for "${insert.tableName}": '
+      '${row.runtimeType}. Expected subclass of TableRow.',
+    );
+  }
+
+  CrdtNode _requireRemoteNode(
+    Map<UuidValue, CrdtNode> remoteNodes,
+    UuidValue uuidNodeId,
+  ) {
+    return remoteNodes[uuidNodeId] ??
+        (throw StateError(
+          'Remote node $uuidNodeId not found for merge.',
+        ));
+  }
+
+  Map<String, Object?> _sanitizeMergeRowData(
+    String tableName,
+    Map<String, Object?> data,
+  ) {
+    final columns = _columnsByTableAndName[tableName];
+    if (columns == null) return {};
+
+    return {
+      for (final MapEntry(key: columnName, value: value) in data.entries)
+        if (columnName != 'id' && columns.containsKey(columnName)) columnName: value,
+    };
+  }
+}

packages/serverpod_offline_sync_client/lib/src/database/recorder.dart

@@ -1,6 +1,9 @@
 import 'package:serverpod_database/serverpod_database.dart';
+import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart';
 import 'package:serverpod_serialization/serverpod_serialization.dart';
 
+import '../crdt/extensions.dart';
+import '../crdt/merge.dart';
 import '../managers/hlc.dart';
 import '../managers/user.dart';
 import '../protocol/protocol.dart';
@@ -8,6 +11,8 @@ import 'database.dart';
 import 'schema.dart';
 import 'session.dart';
 
+part 'merge.dart';
+
 typedef _ReferencingForeignKey = ({
   String childTableName,
   String childColumn,
@@ -129,7 +134,14 @@ class CrdtMutationRecorder {
   /// The list of tables to sync with CRDT.
   final List<Table> syncTables;
 
-  late final _syncTablesNames = syncTables.map((t) => t.tableName).toSet();
+  late final _syncTableByName = {
+    for (final t in syncTables) t.tableName: t,
+  };
+
+  late final Map<String, Set<String>> _syncedTableColumnNamesForMerge = {
+    for (final MapEntry(key: k, value: cols) in _columnsByTableAndName.entries)
+      if (_syncTableByName.containsKey(k)) k: cols.keys.toSet(),
+  };
 
   /// Whether the given table is tracked by CRDT.
   bool isCrdtTracked<T extends TableRow>([Table? table]) {
@@ -140,7 +152,7 @@ class CrdtMutationRecorder {
 
   /// Whether the given table name is tracked by CRDT.
   bool _isCrdtTrackedTableName(String tableName) {
-    return _syncTablesNames.contains(tableName);
+    return _syncTableByName.containsKey(tableName);
   }
 
   /// Insert the CRDT metadata for the inserted rows.
@@ -729,7 +741,7 @@ class CrdtMutationRecorder {
           );
         }
 
-        await _updateDomainRow(tableName, rowId, updates, transaction);
+        await _updateDomainRows(tableName, {rowId}, updates, transaction);
         updatedRowIds.add(rowId);
       }
 
@@ -779,15 +791,6 @@ WHERE d."${_escapeIdentifier(columnName)}" = ${_sqlLiteral(value)}
     };
   }
 
-  Future<void> _updateDomainRow(
-    String tableName,
-    UuidValue rowId,
-    Map<String, Object?> updates,
-    Transaction transaction,
-  ) async {
-    await _updateDomainRows(tableName, {rowId}, updates, transaction);
-  }
-
   Future<void> _updateDomainRows(
     String tableName,
     Set<UuidValue> rowIds,
@@ -797,9 +800,7 @@ WHERE d."${_escapeIdentifier(columnName)}" = ${_sqlLiteral(value)}
     if (rowIds.isEmpty || updates.isEmpty) return;
 
     final assignments = updates.entries
-        .map(
-          (entry) => '"${_escapeIdentifier(entry.key)}" = ${_sqlLiteral(entry.value)}',
-        )
+        .map((e) => '"${_escapeIdentifier(e.key)}" = ${_sqlLiteral(e.value)}')
         .join(', ');
 
     await _db.unsafeExecute(