merge method

@override Future<void> merge(CrdtChangeset changeset)

Merges a CrdtChangeset received from a peer or loaded from storage.

Implementation

@override
Future<void> merge(CrdtChangeset changeset) async {
  var changed = false;
  final touchedTables = <String>{};

  _open.execute('BEGIN');
  try {
    for (final entry in changeset.entries) {
      final table = entry.key;
      for (final record in entry.value) {
        final key = record['key'] as String;
        final incoming = record['hlc'] as Hlc;

        // Advance the local clock; skip a record authored by our own node id
        // (it would be a merge against ourselves).
        try {
          _canonical = _canonical.merge(incoming);
        } catch (_) {
          // DuplicateNodeException / drift — leave the clock as is.
        }

        final existing = _open.select(
          'SELECT hlc FROM crdt_records WHERE collection = ? AND key = ?',
          [table, key],
        );
        if (existing.isNotEmpty &&
            Hlc.parse(existing.first['hlc'] as String) >= incoming) {
          continue;
        }

        _open.execute(
          'INSERT INTO crdt_records (collection, key, value, is_deleted, '
          'hlc, hlc_key, crdt_type) VALUES (?, ?, ?, ?, ?, ?, ?) '
          'ON CONFLICT(collection, key) DO UPDATE SET '
          'value = excluded.value, is_deleted = excluded.is_deleted, '
          'hlc = excluded.hlc, hlc_key = excluded.hlc_key, '
          'crdt_type = excluded.crdt_type',
          [
            table,
            key,
            jsonEncode(record['value']),
            (record['is_deleted'] as bool) ? 1 : 0,
            incoming.toString(),
            _hlcKey(incoming),
            // A peer that predates per-field types (or an in-memory engine)
            // sends no tag; treat it as last-write-wins.
            record['crdt_type'] as String? ?? const LwwRegister().tag,
          ],
        );
        changed = true;
        touchedTables.add(table);
      }
    }
    _open.execute('COMMIT');
  } catch (_) {
    _open.execute('ROLLBACK');
    rethrow;
  }

  if (changed) {
    for (final table in touchedTables) {
      _emit(table);
    }
  }
}