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 {
  // Typed records that must be combined with local state, held back from the
  // inner (last-write-wins) merge and written as fresh local records after
  // it. States are read before the inner merge runs; the sets are disjoint
  // by key, so ordering does not matter.
  final rewrites = <({String table, String key, Object? merged})>[];
  final passthrough = <String, List<CrdtRecord>>{};

  for (final entry in changeset.entries) {
    final table = entry.key;
    final kept = <CrdtRecord>[];
    for (final record in entry.value) {
      final value = record['value'];
      final tag = storedCrdtTag(value);
      final incoming = switch (tag) {
        orSetTag || pnCounterTag when record['is_deleted'] != true => value,
        _ => null,
      };
      if (incoming == null) {
        kept.add(record);
        continue;
      }
      final key = record['key'] as String;
      final local = await _inner.value(table, key);
      if (local == null) {
        // No live local state: apply as a plain record, keeping the
        // incoming HLC so a repeated merge stays a no-op.
        kept.add(record);
        continue;
      }
      final merged = tag == orSetTag
          ? orMerge(local, incoming)
          : pnMerge(local, incoming);
      if (crdtValueEquals(merged, local)) continue; // Nothing new.
      rewrites.add((table: table, key: key, merged: merged));
    }
    if (kept.isNotEmpty) passthrough[table] = kept;
  }

  if (passthrough.isNotEmpty) await _inner.merge(passthrough);
  for (final rewrite in rewrites) {
    await _inner.put(
      rewrite.table,
      rewrite.key,
      rewrite.merged,
      crdtType: crdtTypeFromTag(storedCrdtTag(rewrite.merged)),
    );
  }
}